feat(ingestion/airflow): Add support for mutiple datahub emitter

[treff7es]

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

[codecov]

Codecov Report

Attention: Patch coverage is 85.96491% with 8 lines in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
...plugin/src/datahub_airflow_plugin/hooks/datahub.py	75.00%	4 Missing ⚠️
...n/src/datahub_airflow_plugin/datahub_plugin_v22.py	66.66%	3 Missing ⚠️
...rflow-plugin/src/datahub_airflow_plugin/_config.py	93.33%	1 Missing ⚠️

❌ Your patch status has failed because the patch coverage (80.00%) is below the target coverage (90.00%). You can increase the patch coverage or adjust the target coverage.

Files with missing lines	Coverage Δ
...ingestion/src/datahub/emitter/composite_emitter.py	100.00% <100.00%> (ø)
...rflow-plugin/src/datahub_airflow_plugin/_config.py	98.24% <93.33%> (ø)
...n/src/datahub_airflow_plugin/datahub_plugin_v22.py	60.45% <66.66%> (ø)
...plugin/src/datahub_airflow_plugin/hooks/datahub.py	78.52% <75.00%> (ø)

---

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 dd01c82...64f1d2d. Read the comment docs.

[hsheth2]

I don't think we should have multiple listeners - that would mean that we redo the SQL/lineage extraction work multiple times. imo the listener should remain a singleton.

We discussed the idea of a composite emitter - was there a reason we didn't go that route?

[treff7es]

I don't think we should have multiple listeners - that would mean that we redo the SQL/lineage extraction work multiple times. imo the listener should remain a singleton.

We discussed the idea of a composite emitter - was there a reason we didn't go that route?

I'm sorry, I somehow thought this was a plan you wanted to implement in the future.

[hsheth2]

@treff7es What I was thinking was

• there would continue to be a single Listener class
• listener.emitter -> would return a RestEmitter or KafkaEmitter if there's a single conn id, or a CompositeEmitter / MultiEmitter if there's multiple
• the CompositeEmitter is basically a wrapper class e.g. CompositeEmitter([emitter1, emitter2, ...]). It would only have a couple methods e.g. emit/close which would be pretty simple e.g. for emitter in self._inner_emitters: emitter.emit(...). It would live in metadata-ingestion, not in the airflow plugin.
• The config type / listener init method would be responsible for creating the CompositeEmitter instance
• The rest of the listener's implementation can continue to call listener.emitter.emit(...) and hence would be largely unchanged

[hsheth2]

might also make sense to add a comment explaining that we want the callback to only be called once, and we tie that call to the first emitter. so we should explicitly clarify that errors from other emitters will be suppressed

Suggested change

	callback_called = False
	for emitter in self.emitters:
	if not callback_called:
	emitter.emit(item, callback)
	callback_called = True
	else:
	emitter.emit(item)
	for i, emitter in enumerate(self.emitters):
	emitter.emit(item, callback if i == 0 else None)

[hsheth2]

I think this comment still applies

[hsheth2]

just a thought, not necessary to do: should the DatahubGenericHook be the thing responsible for constructing the composite emitter?

e.g. you can pass it either a string or a list of strings, and it does the right thing