feat(ingest/airbyte): opt in to inactive connection ingestion (#19062)

Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: cubic-dev-ai[bot] <191113872+cubic-dev-ai[bot]@users.noreply.github.com>

Author: 3 people

metadata-ingestion/docs/sources/airbyte/airbyte_post.md

@@ -23,6 +23,13 @@ Module behavior is constrained by source APIs, permissions, and metadata exposed
 
 If ingestion fails, validate credentials, permissions, connectivity, and scope filters first. Then review ingestion logs for source-specific errors and adjust configuration accordingly.
 
+#### Missing Connections
+
+By default only enabled Airbyte connections are ingested. Disabled connections are
+skipped with no warning. If a connection is missing from DataHub, check whether it
+is disabled in Airbyte (or whether its Public API `status` is `"inactive"`). To
+ingest disabled connections as well, set `include_inactive_connections: true`.
+
 #### Authentication Errors
 
 Verify that your OAuth2 client credentials are correct and have not expired. For OSS deployments, confirm the API is reachable at the `/api/public/v1` path prefix.

metadata-ingestion/docs/sources/airbyte/airbyte_recipe.yml

@@ -64,4 +64,4 @@ source:
 sink:
   type: datahub-rest
   config:
-    server: http://localhost:8080
+    server: http://localhost:8080

metadata-ingestion/src/datahub/ingestion/source/airbyte/client.py

@@ -267,7 +267,10 @@ def list_workspaces(
         return [AirbyteWorkspacePartial.model_validate(w) for w in workspaces_data]
 
     def list_connections(
-        self, workspace_id: str, pattern: Optional[AllowDenyPattern] = None
+        self,
+        workspace_id: str,
+        pattern: Optional[AllowDenyPattern] = None,
+        include_inactive: bool = False,
     ) -> List[AirbyteConnectionPartial]:
         self._check_auth_before_request()
         params = {API_QUERY_WORKSPACE_ID: workspace_id}
@@ -279,16 +282,17 @@ def list_connections(
             )
         )
 
-        active_connections = [
-            conn
-            for conn in connections
-            if conn.get(API_FIELD_STATUS) != API_STATUS_INACTIVE
-        ]
+        if not include_inactive:
+            connections = [
+                conn
+                for conn in connections
+                if conn.get(API_FIELD_STATUS) != API_STATUS_INACTIVE
+            ]
 
         if pattern:
-            active_connections = apply_pattern(active_connections, pattern)
+            connections = apply_pattern(connections, pattern)
 
-        return [AirbyteConnectionPartial.model_validate(c) for c in active_connections]
+        return [AirbyteConnectionPartial.model_validate(c) for c in connections]
 
     def list_jobs(
         self,

metadata-ingestion/src/datahub/ingestion/source/airbyte/config.py

@@ -263,6 +263,15 @@ class AirbyteSourceConfig(
         "Use this to override platform details for specific destinations.",
     )
 
+    include_inactive_connections: bool = Field(
+        default=False,
+        description=(
+            "Also ingest connections that are disabled in Airbyte. "
+            "By default, connections reported as inactive are skipped with no "
+            "warning, even if they still appear healthy in the Airbyte UI."
+        ),
+    )
+
     include_statuses: bool = Field(
         default=True,
         description="Whether to ingest run statuses",

metadata-ingestion/src/datahub/ingestion/source/airbyte/source.py

@@ -281,6 +281,7 @@ def _get_pipelines(self) -> Iterable[AirbytePipelineInfo]:
                 for connection in self.client.list_connections(
                     workspace.workspace_id,
                     pattern=self.source_config.connection_pattern,
+                    include_inactive=self.source_config.include_inactive_connections,
                 ):
                     try:
                         if (

metadata-ingestion/tests/unit/airbyte/test_airbyte_client.py

@@ -185,6 +185,36 @@ def test_list_connections(self, mock_paginate_results):
             result_key="data",
         )
 
+    @patch("datahub.ingestion.source.airbyte.client.AirbyteOSSClient._paginate_results")
+    def test_list_connections_skips_inactive_by_default(self, mock_paginate_results):
+        mock_paginate_results.return_value = [
+            {
+                "connectionId": "active-id",
+                "name": "Active Connection",
+                "sourceId": "source-id-1",
+                "destinationId": "destination-id-1",
+                "status": "active",
+            },
+            {
+                "connectionId": "inactive-id",
+                "name": "Inactive Connection",
+                "sourceId": "source-id-2",
+                "destinationId": "destination-id-2",
+                "status": "inactive",
+            },
+        ]
+        config = AirbyteClientConfig(
+            deployment_type=AirbyteDeploymentType.OPEN_SOURCE,
+            host_port="http://localhost:8000",
+        )
+        client = AirbyteOSSClient(config)
+
+        connections = client.list_connections("workspace-id-1")
+        assert [c.connection_id for c in connections] == ["active-id"]
+
+        connections = client.list_connections("workspace-id-1", include_inactive=True)
+        assert [c.connection_id for c in connections] == ["active-id", "inactive-id"]
+
     @patch("datahub.ingestion.source.airbyte.client.AirbyteOSSClient._make_request")
     def test_http_error_handling(self, mock_make_request):
         mock_make_request.side_effect = requests.exceptions.HTTPError(
@@ -226,7 +256,9 @@ def list_sources(self, workspace_id, pattern=None):
             def list_destinations(self, workspace_id, pattern=None):
                 return []
 
-            def list_connections(self, workspace_id, pattern=None):
+            def list_connections(
+                self, workspace_id, pattern=None, include_inactive=False
+            ):
                 return []
 
         class IncompleteClient(AirbyteBaseClient):

metadata-ingestion/tests/unit/airbyte/test_source_workspaces.py

@@ -122,7 +122,9 @@ def test_get_pipelines(mock_create_client, mock_ctx, mock_client):
 
     mock_client.list_workspaces.assert_called_once()
     mock_client.list_connections.assert_called_once_with(
-        "workspace-1", pattern=AllowDenyPattern.allow_all()
+        "workspace-1",
+        pattern=AllowDenyPattern.allow_all(),
+        include_inactive=False,
     )
     mock_client.get_connection.assert_called_once_with("connection-1")
     mock_client.get_source.assert_called_once_with("source-1")
@@ -191,6 +193,73 @@ def test_get_pipelines_with_filters(mock_create_client, mock_ctx, mock_client):
     assert len(pipelines) == 0
 
 
+@patch("datahub.ingestion.source.airbyte.source.create_airbyte_client")
+def test_get_pipelines_include_inactive_connections(
+    mock_create_client, mock_ctx, mock_client
+):
+    mock_create_client.return_value = mock_client
+    config = AirbyteSourceConfig(
+        deployment_type=AirbyteDeploymentType.OPEN_SOURCE,
+        host_port="http://localhost:8000",
+        platform_instance="test-instance",
+        include_inactive_connections=True,
+    )
+    source = AirbyteSource(config, mock_ctx)
+
+    workspace = AirbyteWorkspacePartial(
+        workspace_id="workspace-1",
+        name="Test Workspace",
+    )
+    connection = AirbyteConnectionPartial(
+        connection_id="inactive-connection-1",
+        name="Disabled Connection",
+        source_id="source-1",
+        destination_id="destination-1",
+        status="inactive",
+    )
+    source_model = AirbyteSourcePartial(
+        source_id="source-1",
+        name="Test Source",
+        source_type="postgres",
+        source_definition_id="source-def-1",
+        workspace_id="workspace-1",
+        configuration={"host": "localhost", "port": 5432},
+    )
+    destination = AirbyteDestinationPartial(
+        destination_id="destination-1",
+        name="Test Destination",
+        destination_type="postgres",
+        destination_definition_id="dest-def-1",
+        workspace_id="workspace-1",
+        configuration={"host": "localhost", "port": 5432},
+    )
+
+    mock_client.list_workspaces.return_value = [workspace]
+    mock_client.list_connections.return_value = [connection]
+    mock_client.get_connection.return_value = connection
+    mock_client.get_source.return_value = source_model
+    mock_client.get_destination.return_value = destination
+
+    pipelines = list(source._get_pipelines())
+
+    assert len(pipelines) == 1
+    assert isinstance(pipelines[0], AirbytePipelineInfo)
+    assert pipelines[0].workspace.workspace_id == "workspace-1"
+    assert pipelines[0].connection.connection_id == "inactive-connection-1"
+    assert pipelines[0].connection.status == "inactive"
+    assert pipelines[0].source.source_id == "source-1"
+    assert pipelines[0].destination.destination_id == "destination-1"
+
+    mock_client.list_connections.assert_called_once_with(
+        "workspace-1",
+        pattern=AllowDenyPattern.allow_all(),
+        include_inactive=True,
+    )
+    mock_client.get_connection.assert_called_once_with("inactive-connection-1")
+    mock_client.get_source.assert_called_once_with("source-1")
+    mock_client.get_destination.assert_called_once_with("destination-1")
+
+
 @patch("datahub.ingestion.source.airbyte.source.create_airbyte_client")
 @patch(
     "datahub.ingestion.source.airbyte.source.AirbyteSource._create_lineage_workunits"