yujunjun/OpenMetadata
1
0
Fork 0
mirror of https://github.com/open-metadata/OpenMetadata.git synced 2025-08-07 16:48:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix Usage from Bulk Sink (#3354)

Browse Source 
* Fix Usage from Bulk Sink

* Closing stage in worklow once bulk sink starts
This commit is contained in:
Ayush Shah 2022-03-10 17:12:42 +05:30 committed by GitHub
parent 794cf64787
commit 4f1ef18c2b
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 1 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
ingestion/src/metadata/ingestion/api/workflow.py
View File

@ -163,14 +163,13 @@ class Workflow:
self.sink.write_record(processed_record) self.sink.write_record(processed_record)
self.report["sink"] = self.sink.get_status().as_obj() self.report["sink"] = self.sink.get_status().as_obj()
if hasattr(self, "bulk_sink"): if hasattr(self, "bulk_sink"):
self.stage.close()
self.bulk_sink.write_records() self.bulk_sink.write_records()
self.report["Bulk_Sink"] = self.bulk_sink.get_status().as_obj() self.report["Bulk_Sink"] = self.bulk_sink.get_status().as_obj()
def stop(self): def stop(self):
if hasattr(self, "processor"): if hasattr(self, "processor"):
self.processor.close() self.processor.close()
if hasattr(self, "stage"):
self.stage.close()
if hasattr(self, "bulk_sink"): if hasattr(self, "bulk_sink"):
self.bulk_sink.close() self.bulk_sink.close()
if hasattr(self, "sink"): if hasattr(self, "sink"):