yujunjun/datahub
2
0
Fork 0
mirror of https://github.com/datahub-project/datahub.git synced 2025-11-14 18:31:59 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(ingest): DataProcessInstance.emit_process_end() ignored start_timestamp_millis (#10539)

Browse Source
This commit is contained in:
Oleksandr Baltian 2024-05-24 20:34:00 +02:00 committed by GitHub
parent 47ba6d9395
commit c308a6ff45
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 5 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
metadata-ingestion/src/datahub/api/entities/dataprocess/dataprocess_instance.py
View File

@ -176,6 +176,8 @@ class DataProcessInstance:
:param result: (InstanceRunResult) the result of the run :param result: (InstanceRunResult) the result of the run
:param result_type: (string) It identifies the system where the native result comes from like Airflow, Azkaban :param result_type: (string) It identifies the system where the native result comes from like Airflow, Azkaban
:param attempt: (int) the attempt number of this execution :param attempt: (int) the attempt number of this execution
:param start_timestamp_millis: (Optional[int]) the start time of the execution in milliseconds
""" """
mcp = MetadataChangeProposalWrapper( mcp = MetadataChangeProposalWrapper(
entityUrn=str(self.urn), entityUrn=str(self.urn),
@ -215,12 +217,15 @@ class DataProcessInstance:
:param result_type: (string) It identifies the system where the native result comes from like Airflow, Azkaban :param result_type: (string) It identifies the system where the native result comes from like Airflow, Azkaban
:param attempt: (int) the attempt number of this execution :param attempt: (int) the attempt number of this execution
:param callback: (Optional[Callable[[Exception, str], None]]) the callback method for KafkaEmitter if it is used :param callback: (Optional[Callable[[Exception, str], None]]) the callback method for KafkaEmitter if it is used
:param start_timestamp_millis: (Optional[int]) the start time of the execution in milliseconds
""" """
for mcp in self.end_event_mcp( for mcp in self.end_event_mcp(
end_timestamp_millis=end_timestamp_millis, end_timestamp_millis=end_timestamp_millis,
result=result, result=result,
result_type=result_type, result_type=result_type,
attempt=attempt, attempt=attempt,
start_timestamp_millis=start_timestamp_millis,
): ):
self._emit_mcp(mcp, emitter, callback) self._emit_mcp(mcp, emitter, callback)