fix(ingestion/fivetran): Fix fivetran get connector jobs bug (#9975)

datahub-project · Mar 5, 2024 · 4d053f1 · 4d053f1
1 parent a0efb9d
commit 4d053f1
Show file tree

Hide file tree

Showing 2 changed files with 13 additions and 1 deletion.
diff --git a/metadata-ingestion/src/datahub/ingestion/source/fivetran/fivetran_log_api.py b/metadata-ingestion/src/datahub/ingestion/source/fivetran/fivetran_log_api.py
@@ -132,7 +132,10 @@ def _get_jobs_list(self, connector_id: str) -> List[Job]:
                 # If no sync-end event log for this sync id that means sync is still in progress
                 continue
 
-            message_data = json.loads(sync_end_logs[sync_id][Constant.MESSAGE_DATA])
+            message_data = sync_end_logs[sync_id][Constant.MESSAGE_DATA]
+            if message_data is None:
+                continue
+            message_data = json.loads(message_data)
             if isinstance(message_data, str):
                 # Sometimes message_data contains json string inside string
                 # Ex: '"{\"status\":\"SUCCESSFUL\"}"'

diff --git a/metadata-ingestion/tests/integration/fivetran/test_fivetran.py b/metadata-ingestion/tests/integration/fivetran/test_fivetran.py
@@ -95,6 +95,10 @@ def default_query_results(
                 "time_stamp": datetime.datetime(2023, 10, 3, 14, 35, 55, 401000),
                 "sync_id": "63c2fc85-600b-455f-9ba0-f576522465be",
             },
+            {
+                "time_stamp": datetime.datetime(2023, 10, 3, 14, 37, 5, 403000),
+                "sync_id": "e773e1e9-c791-46f4-894f-8ch9b3dfc832",
+            },
         ]
     elif query == fivetran_log_query.get_sync_end_logs_query("calendar_elected"):
         return [
@@ -113,6 +117,11 @@ def default_query_results(
                 "sync_id": "63c2fc85-600b-455f-9ba0-f576522465be",
                 "message_data": '"{\\"reason\\":\\"java.lang.RuntimeException: FATAL: too many connections for role \\\\\\"hxwraqld\\\\\\"\\",\\"taskType\\":\\"reconnect\\",\\"status\\":\\"FAILURE_WITH_TASK\\"}"',
             },
+            {
+                "time_stamp": datetime.datetime(2023, 10, 3, 14, 37, 35, 478000),
+                "sync_id": "e773e1e9-c791-46f4-894f-8ch9b3dfc832",
+                "message_data": None,
+            },
         ]
     # Unreachable code
     raise Exception(f"Unknown query {query}")