fix(ingestion): fix stateful ingestion for GCS source

[josges]

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

Checklist

• The PR conforms to DataHub's Contributing Guideline (particularly Commit Message Format)
• Links to related issues (if applicable)
• Tests for the changes have been added/updated (if applicable)
• Docs related to the changes have been added/updated (if applicable). If a new feature has been added a Usage Guide has been added for the same.
• For any breaking change/potential downtime/deprecation/big changes an entry has been made in Updating DataHub

[sgomezvillamor]

I think this constant is available

Suggested change

	self.platform: str = "gcs"
	self.platform: str = PLATFORM_GCS

[josges]

Changed, thank you!

[sgomezvillamor]

You may move these two lines inside create_equivalent_s3_source

Also, how is s3_ctx.pipeline_name = None is required?
Note pipeline_name is required when stateful_ingestion is enabled.

[josges]

Thanks a lot for your feedback!

Without this, running stateful ingestion leads to IndexError: Checkpointing provider DatahubIngestionCheckpointingProvider already registered.. This is because StatefulIngestionSourceBase is initialized twice, first for GCSSource and then in create_equivalent_s3_source with S3Source. This can also be verified by the change I did in the unit test.
I think, stateful ingestion was always meant to be switched off for the equivalent S3 source object, because in create_equivalent_s3_config, DataLakeSourceConfig is initialized without stateful_ingestion, which implicitly means stateful_ingestion=None. Since pipeline_name is set, the logic in StateProviderWrapper leads to stateful ingestion being switched on, which leads to the error.

I put the lines in the function, as suggested.