refactor

datahub-project · Dec 13, 2024 · 1748219 · 1748219
1 parent d0e1f41
commit 1748219
Showing 1 changed file with 24 additions and 13 deletions.
diff --git a/metadata-ingestion/src/datahub/ingestion/source/kafka/kafka_connect.py b/metadata-ingestion/src/datahub/ingestion/source/kafka/kafka_connect.py
@@ -1171,19 +1171,15 @@ def get_connectors_manifest(self) -> List[ConnectorManifest]:
         payload = connector_response.json()
 
         for connector_name in payload:
-            try:
-                connector_url = f"{self.config.connect_uri}/connectors/{connector_name}"
-                connector_response = self.session.get(connector_url)
-                connector_response.raise_for_status()
-            except Exception as e:
-                self.report.warning(
-                    "Failed to get connector details", connector_name, exc=e
-                )
-                continue
-            manifest = connector_response.json()
-            connector_manifest = ConnectorManifest(**manifest)
-            if not self.config.connector_patterns.allowed(connector_manifest.name):
-                self.report.report_dropped(connector_manifest.name)
+            connector_url = f"{self.config.connect_uri}/connectors/{connector_name}"
+            connector_manifest = self._get_connector_manifest(
+                connector_name, connector_url
+            )
+            if (
+                connector_manifest is None
+                or not self.config.connector_patterns.allowed(connector_manifest.name)
+            ):
+                self.report.report_dropped(connector_name)
                 continue
 
             if self.config.provided_configs:
@@ -1277,6 +1273,21 @@ def get_connectors_manifest(self) -> List[ConnectorManifest]:
 
         return connectors_manifest
 
+    def _get_connector_manifest(
+        self, connector_name: str, connector_url: str
+    ) -> Optional[ConnectorManifest]:
+        try:
+            connector_response = self.session.get(connector_url)
+            connector_response.raise_for_status()
+        except Exception as e:
+            self.report.warning(
+                "Failed to get connector details", connector_name, exc=e
+            )
+            return None
+        manifest = connector_response.json()
+        connector_manifest = ConnectorManifest(**manifest)
+        return connector_manifest
+
     def _get_connector_tasks(self, connector_name: str) -> dict:
         try:
             response = self.session.get(