Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

fix(ingest): pick topics from config for sink connector #12535

Merged
merged 7 commits into from
Feb 13, 2025
Merged

fix(ingest): pick topics from config for sink connector #12535

merged 7 commits into from
Feb 13, 2025

Conversation

mayurinehate
Copy link
Collaborator

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

@github-actions github-actions bot added the ingestion PR or Issue related to the ingestion of metadata label Feb 3, 2025
@datahub-cyborg datahub-cyborg bot added the needs-review Label for PRs that need review from a maintainer. label Feb 3, 2025
@codecov Codecov
Copy link

codecov bot commented Feb 3, 2025 •
edited
Loading

Codecov Report

Attention: Patch coverage is 90.90909% with 3 lines in your changes missing coverage. Please review.

Files with missing lines Patch % Lines
...ub/ingestion/source/kafka_connect/kafka_connect.py 89.28% 3 Missing ⚠️
Files with missing lines Coverage Δ
...c/datahub/ingestion/source/kafka_connect/common.py 99.02% <100.00%> (ø)
.../ingestion/source/kafka_connect/sink_connectors.py 64.11% <100.00%> (ø)
...ngestion/source/kafka_connect/source_connectors.py 81.05% <100.00%> (ø)
...ub/ingestion/source/kafka_connect/kafka_connect.py 91.52% <89.28%> (-0.64%) ⬇️

... and 35 files with indirect coverage changes

Continue to review full report in Codecov by Sentry.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 7472c53...d644055. Read the comment docs.

hsheth2
hsheth2 reviewed Feb 5, 2025
View reviewed changes
Copy link
Collaborator

@hsheth2 hsheth2 left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

can we add some tests for this?

@datahub-cyborg datahub-cyborg bot added pending-submitter-response Issue/request has been reviewed but requires a response from the submitter and removed needs-review Label for PRs that need review from a maintainer. labels Feb 5, 2025
@datahub-cyborg datahub-cyborg bot added needs-review Label for PRs that need review from a maintainer. and removed pending-submitter-response Issue/request has been reviewed but requires a response from the submitter labels Feb 5, 2025
@mayurinehate
Copy link
Collaborator Author

Updated test here

Without changes, other than config - stale topic shows up and after changes, it does not.
Also this tests the logic to extract topics from topics config in sink.

@mayurinehate mayurinehate marked this pull request as draft February 5, 2025 07:51
@mayurinehate mayurinehate marked this pull request as ready for review February 11, 2025 14:11
hsheth2
hsheth2 approved these changes Feb 11, 2025
View reviewed changes
metadata-ingestion/src/datahub/ingestion/source/kafka_connect/kafka_connect.py
@@ -121,7 +121,9 @@ def get_connectors_manifest(self) -> Iterable[ConnectorManifest]:
connector_manifest.config, self.config.provided_configs
)
connector_manifest.url = connector_url
connector_manifest.topic_names = self._get_connector_topics(connector_name)
connector_manifest.topic_names = self._get_connector_topics(
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit - ideally when calling this method, we'd use kwargs e.g. self._get_connector_topics(connector=..., type=...)

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ideally, this should become a lint checker
Unfortunately, this is not supported by ruff yet astral-sh/ruff#3269

metadata-ingestion/src/datahub/ingestion/source/kafka_connect/kafka_connect.py
class SinkTopicFilter:
"""Helper class to filter Kafka Connect topics based on configuration."""

def filter_stale_topics(
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

can you add a comment explaining why we need this

something like kafka-connect's API returns an over-populated list of topics. if a topic was ever used by a connector, it will be returned, even if it is no longer used. to remove these stale topics from the list, we double-check the list returned by the API against the sink connector's config

metadata-ingestion/src/datahub/ingestion/source/kafka_connect/sink_connectors.py
@@ -175,7 +175,7 @@ class BigQuerySinkConnector(BaseConnector):
class BQParser:
project: str
target_platform: str
sanitizeTopics: str
sanitizeTopics: bool
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

was the type annotation here previously wrong?

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes

@datahub-cyborg datahub-cyborg bot added pending-submitter-merge and removed needs-review Label for PRs that need review from a maintainer. labels Feb 11, 2025
sgomezvillamor
sgomezvillamor reviewed Feb 11, 2025
View reviewed changes
metadata-ingestion/tests/integration/kafka-connect/test_kafka_connect.py
@@ -682,3 +687,43 @@ def test_kafka_connect_bigquery_sink_ingest(
golden_path=test_resources_dir / "kafka_connect_bigquery_sink_mces_golden.json",
ignore_paths=[],
)


Copy link
Contributor

@sgomezvillamor sgomezvillamor Feb 11, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

while it may be obvious, the case sink_config missing both topics and topics.regex keys may be added for completeness

@mayurinehate mayurinehate merged commit e2a98d2 into datahub-project:master Feb 13, 2025
189 of 190 checks passed
ttekampe pushed a commit to ttekampe/datahub that referenced this pull request Feb 14, 2025
@mayurinehate
fix(ingest): pick topics from config for sink connector (datahub-proj…
d10f0a0 
…ect#12535)
ksrinath pushed a commit to ksrinath/datahub that referenced this pull request Feb 14, 2025
@mayurinehate @ksrinath
fix(ingest): pick topics from config for sink connector (datahub-proj…
072cd53 
…ect#12535)
shirshanka pushed a commit to shirshanka/datahub that referenced this pull request Mar 3, 2025
@mayurinehate @shirshanka
fix(ingest): pick topics from config for sink connector (datahub-proj…
dae4f58 
…ect#12535)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Reviewers

@sgomezvillamor sgomezvillamor sgomezvillamor left review comments

@hsheth2 hsheth2 hsheth2 approved these changes

Assignees
No one assigned
Labels
ingestion PR or Issue related to the ingestion of metadata pending-submitter-merge
Projects
None yet
Milestone
No milestone
Development

Successfully merging this pull request may close these issues.
3 participants
@mayurinehate @sgomezvillamor @hsheth2