fix(ingestion): fix stateful ingestion for GCS source

Remove pipeline name before passing context to equivalent s3 source to avoid error "Checkpointing provider DatahubIngestionCheckpointingProvider already registered."

metadata-ingestion/src/datahub/ingestion/source/gcs/gcs_source.py

@@ -1,3 +1,4 @@
+import copy
 import logging
 from typing import Dict, Iterable, List, Optional
 from urllib.parse import unquote
@@ -88,7 +89,10 @@ def __init__(self, config: GCSSourceConfig, ctx: PipelineContext):
         super().__init__(config, ctx)
         self.config = config
         self.report = GCSSourceReport()
-        self.s3_source = self.create_equivalent_s3_source(ctx)
+        self.platform: str = "gcs"
+        s3_ctx = copy.deepcopy(ctx)
+        s3_ctx.pipeline_name = None
+        self.s3_source = self.create_equivalent_s3_source(s3_ctx)
 
     @classmethod
     def create(cls, config_dict, ctx):

metadata-ingestion/tests/unit/test_gcs_source.py

@@ -1,13 +1,17 @@
+from unittest import mock
+
 import pytest
 from pydantic import ValidationError
 
 from datahub.ingestion.api.common import PipelineContext
+from datahub.ingestion.graph.client import DataHubGraph
 from datahub.ingestion.source.data_lake_common.data_lake_utils import PLATFORM_GCS
 from datahub.ingestion.source.gcs.gcs_source import GCSSource
 
 
 def test_gcs_source_setup():
-    ctx = PipelineContext(run_id="test-gcs")
+    graph = mock.MagicMock(spec=DataHubGraph)
+    ctx = PipelineContext(run_id="test-gcs", graph=graph, pipeline_name="test-gcs")
 
     # Baseline: valid config
     source: dict = {
@@ -18,6 +22,7 @@ def test_gcs_source_setup():
             }
         ],
         "credential": {"hmac_access_id": "id", "hmac_access_secret": "secret"},
+        "stateful_ingestion": {"enabled": "true"},
     }
     gcs = GCSSource.create(source, ctx)
     assert gcs.s3_source.source_config.platform == PLATFORM_GCS