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): better handling around sink errors #9003

Merged
merged 3 commits into from
Oct 25, 2023
Merged

fix(ingest): better handling around sink errors #9003

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
f4d3c1a
fix(ingest): better handling around sink errors
hsheth2 Oct 13, 2023
b9b1437
Merge branch 'master' into kafka-sink-errors
hsheth2 Oct 24, 2023
0c5e832
Merge branch 'master' into kafka-sink-errors
hsheth2 Oct 25, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 9 additions & 1 deletion metadata-ingestion/src/datahub/ingestion/run/pipeline.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -390,7 +390,15 @@ def run(self) -> None:
record_envelopes = self.extractor.get_records(wu)
for record_envelope in self.transform(record_envelopes):
if not self.dry_run:
self.sink.write_record_async(record_envelope, callback)
try:
self.sink.write_record_async(
record_envelope, callback
)
except Exception as e:
# In case the sink's error handling is bad, we still want to report the error.
self.sink.report.report_failure(
f"Failed to write record: {e}"
Copy link
Contributor

Choose a reason for hiding this comment

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

Can log stack trace? I'm a bit concerned about losing information where the error came.

Copy link
Collaborator Author

Choose a reason for hiding this comment

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

this is mainly a fallback mechanism if the source doesn't catch the error elsewhere - given that the error handling in the kafka sink has also improved, I think it's ok

)

except RuntimeError:
raise
Expand Down
33 changes: 13 additions & 20 deletions metadata-ingestion/src/datahub/ingestion/sink/datahub_kafka.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,6 @@
MetadataChangeEvent,
MetadataChangeProposal,
)
from datahub.metadata.schema_classes import MetadataChangeProposalClass


class KafkaSinkConfig(KafkaEmitterConfig):
Expand Down Expand Up @@ -58,27 +57,21 @@ def write_record_async(
],
write_callback: WriteCallback,
) -> None:
record = record_envelope.record
if isinstance(record, MetadataChangeEvent):
self.emitter.emit_mce_async(
callback = _KafkaCallback(
self.report, record_envelope, write_callback
).kafka_callback
try:
record = record_envelope.record
self.emitter.emit(
record,
callback=_KafkaCallback(
self.report, record_envelope, write_callback
).kafka_callback,
)
elif isinstance(
record, (MetadataChangeProposalWrapper, MetadataChangeProposalClass)
):
self.emitter.emit_mcp_async(
record,
callback=_KafkaCallback(
self.report, record_envelope, write_callback
).kafka_callback,
)
else:
raise ValueError(
f"The datahub-kafka sink only supports MetadataChangeEvent/MetadataChangeProposal[Wrapper] classes, not {type(record)}"
callback=callback,
)
except Exception as err:
# In case we throw an exception while trying to emit the record,
# catch it and report the failure. This might happen if the schema
# registry is down or otherwise misconfigured, in which case we'd
# fail when serializing the record.
callback(err, f"Failed to write record: {err}")

def close(self) -> None:
self.emitter.flush()
Loading