Add a DagRun and TaskInstance event listener under the Kafka provider

[xBis7]

---

This PR is adding a listener under the Kafka provider which provides an implementation for the DagRun and TaskInstance state change event hooks.

The listener publishes a message to a pre-existing Kafka topic for every event. Each message has some extra metadata for the dag or task that the event belongs to.

DagRun events and TaskInstance events are separated and guarded behind different config flags. Both flags are disabled by default because each listener adds load to Airflow that users might not want.

The listener expects that the user has already created the topic and then defines it in the listener's config section. In case the topic doesn't exist, the listener doesn't fail. Instead it logs a warning and checks for the topic existence again after a configured interval.

Users can also filter events for dag runs based on dag_id and for tasks based on dag_id + task_id. For example, someone could choose to get only dag run state events for a particular dag and nothing else.

By adding a listener to the provider we can have dags consume from the Kafka topic and react to messages. When a particular message arrives, a deferred task could be triggered. This could also work for dags across multiple Airflow installations, all sharing the common topic. For that reason, I also added a source config key, so that users can distinguish where the messages came from.

The listener follows the same format as the openlineage one

https://github.com/apache/airflow/blob/main/providers/openlineage/src/airflow/providers/openlineage/plugins/listener.py

Testing

I've added unit tests and also a simple integration test that uses an actual broker. I've also tested the changes manually.

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)
  Claude Code, Opus 4.7

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[vikramkoka]

I am confused by the use case here.

Shouldn't this be an Asset Watcher?

[xBis7]

Shouldn't this be an Asset Watcher?

@vikramkoka No but it could be used together with an Asset Watcher. An Asset Watcher is essentially on the consumer side of events while this PR is on the producer side.

The new listener produces a message to a Kafka topic for every event state change. We can consume the messages from the topic either using the existing consumer hook from the Kafka provider or using an Asset Watcher.

For example, team A with Airflow installation A triggers dag1 and has enabled the listener which publishes events for every state change to a Kafka topic

1. dag_run.dag1.running
2. task1.running
3. task1.success
4. task2.running
5. task2.success
6. dag_run.dag1.success

Team B with Airflow installation B, is monitoring the same kafka topic and has a deferred task that waits to run after it consumes the message task1.success. The message consumption can happen using an Asset Watcher.

[jason810496]

Nice, LGTM overall.

this PR is on the producer side.

Yes. I think this is a valid feature for Kafka provider to support out of the box.

[xBis7]

Hey @jason810496, I've updated the PR. It took me more time than originally planned but I needed to try a few different approaches and test things thoroughly. I think that this last revision is better.

I added two new options in the listener config

1. set the kafka connection name
2. set a json dictionary with the kafka config

The producer, consumer and admin client hooks, all require an airflow connection to be initialized. The airflow connection carries an extra param JSON that holds all the necessary configuration. This configuration can also carry security info.

Initially I tried to workaround it by defining bootstrap.servers in the listener config, so that users can initialize a producer without creating a connection.

Confluent kafka producer accepts the config as a dictionary. But the airflow connection json is a string, which has limitation like not being able to pass standard callables like on_delivery or oauth_cb, etc. without having to hardcode them in the hook or the caller.

I added an optional parameter to the base class, so that hook callers can pass the config dictionary directly to the hook without using a connection. The param is optional for back-compat reasons. Also, I added a new method that interprets the config callables whether they are part of the connection or the dictionary. It accepts dot-path strings pointing to locally installed packages.

If a user has a connection but has also set a config key in the dict param, the dict value overrides the value from the connection.

The changes are more extensive than originally planned but I think that they improve the way the hooks are configured greatly.

Please take a look and let me know how it looks. I also updated all the tests.

[jason810496]

Nice, thanks. LGMT overall.

[jason810496]

Nice! LGTM, we can merge once the CI in main branch fix.

[xBis7]

The CI is green. It should be good to get merged.

[xBis7]

Thank you @shahar1 @jason810496 for the reviews and help!

[vincbeck]

I went through the PR and here is what I think. This is not a replacement or alternative of asset watcher, they serve 2 different purposes:

• Asset watcher. Schedule/trigger dags based on external event such as a message in Kafka topic
• This PR. Emit events to kafka topic on dag run and task instance state change

The name listener is definitely confusing so we should change that. Here are some idea on how we could call it:

• Event publisher
• Event exporter
• Event forwarder
• Event reporter

Not related but side comment on this one. If we know this is a business case needed by users we should make it a Airflow feature (with providers being able to hook their implementation such as Kafka) as opposed to a Kafka feature.

@vikramkoka

[xBis7]

@vincbeck Thank you for the comment! I've started working on renaming the listener to Event Producer as a suggestion from @shahar1. Event Publisher is also accurate but producer is the term generally used by Kafka, and so it made sense to me. If you don't mind, I'll move forward with that.

OpenLineage and Informatica also have a similar listener plugin

https://github.com/apache/airflow/blob/main/providers/openlineage/src/airflow/providers/openlineage/plugins/listener.py

https://github.com/apache/airflow/blob/main/providers/informatica/src/airflow/providers/informatica/plugins/listener.py

Also see

https://github.com/apache/airflow/blob/main/airflow-core/src/airflow/listeners/listener.py

https://github.com/apache/airflow/blob/main/airflow-core/src/airflow/listeners/spec/dagrun.py

https://github.com/apache/airflow/blob/main/shared/listeners/src/airflow_shared/listeners/spec/taskinstance.py

[vincbeck]

Oh I see, interesting. It is kind of weird to me to expect deployment manager to have to install a plugin to set that up but that's another discussion we can leave for another time :)