Configuration Reference¶

This page contains the list of all available Airflow configurations for the apache-airflow-providers-apache-kafka provider that can be set in the airflow.cfg file or using environment variables.

Note

For more information see Setting Configuration Options.

[apache_kafka]¶

Common settings for the Apache Kafka Provider.

callback_allowlist¶

Comma-separated list of callbacks that may be resolved from a Kafka connection extra into confluent-kafka callback options (error_cb, throttle_cb, stats_cb, log_cb, oauth_cb, on_commit). Each entry is the full importable path of the callback itself — module plus attribute, e.g. my_company.kafka.auth.oauth_cb — matched exactly against the dotted-path string on the connection extra (a bare module such as my_company.kafka.auth won’t lead to authorization of the callables inside it). This is enforced for security reasons, to prevent malicious callbacks from being executed: only paths listed here are resolved, and any other string-valued callback is refused. Empty (default) disables string-valued callbacks entirely. Managed authentication (Amazon MSK IAM, Google Managed Kafka) does not rely on this and is unaffected.

Type:: string
Default:: ''
Environment Variable:: AIRFLOW__APACHE_KAFKA__CALLBACK_ALLOWLIST
Example:: my_company.kafka.auth.oauth_cb

[kafka_event_producer]¶

Settings for the Kafka event producer plugin that publishes Airflow DagRun and TaskInstance state-change events to a Kafka topic.

dag_run_dag_id_allowlist¶

Comma-separated glob patterns. When set, DagRun events are only emitted for dag_ids matching at least one pattern. Empty = all dags.

Type:: string
Default:: ''
Environment Variable:: AIRFLOW__KAFKA_EVENT_PRODUCER__DAG_RUN_DAG_ID_ALLOWLIST
Example:: demo_*,test_dag1

dag_run_dag_id_denylist¶

Comma-separated glob patterns. DagRun events for dag_ids matching any pattern are skipped. Deny takes precedence over allow.

Type:: string
Default:: ''
Environment Variable:: AIRFLOW__KAFKA_EVENT_PRODUCER__DAG_RUN_DAG_ID_DENYLIST
Example:: demo_*

dag_run_events_enabled¶

Publish DagRun state-change events (dag_run.running, dag_run.success, dag_run.failed). When False the DagRun listener is not registered.

Type:: boolean
Default:: False
Environment Variable:: AIRFLOW__KAFKA_EVENT_PRODUCER__DAG_RUN_EVENTS_ENABLED

kafka_config_id¶

Airflow connection used to build the plugin’s Kafka producer. When unset, the producer hook falls back to its default connection (kafka_default).

Type:: string
Default:: ''
Environment Variable:: AIRFLOW__KAFKA_EVENT_PRODUCER__KAFKA_CONFIG_ID
Example:: kafka_default

source¶

Identifier added to every emitted message under the source field so consumers can distinguish Airflow installations that share the same topic. When unset, falls back to the hostname of the Airflow component that emits the event (scheduler, worker, etc.).

Type:: string
Default:: ''
Environment Variable:: AIRFLOW__KAFKA_EVENT_PRODUCER__SOURCE
Example:: af-prod-eu

task_instance_dag_id_allowlist¶

Comma-separated glob patterns. When set, TaskInstance events are only emitted for dag_ids matching at least one pattern. Empty = all dags.

Type:: string
Default:: ''
Environment Variable:: AIRFLOW__KAFKA_EVENT_PRODUCER__TASK_INSTANCE_DAG_ID_ALLOWLIST
Example:: demo_*,test_dag1

task_instance_dag_id_denylist¶

Comma-separated glob patterns. TaskInstance events for dag_ids matching any pattern are skipped. Deny takes precedence over allow.

Type:: string
Default:: ''
Environment Variable:: AIRFLOW__KAFKA_EVENT_PRODUCER__TASK_INSTANCE_DAG_ID_DENYLIST
Example:: demo_*

task_instance_events_enabled¶

Publish TaskInstance state-change events (task_instance.running, task_instance.success, task_instance.failed, task_instance.skipped). When False the TaskInstance listener is not registered.

Type:: boolean
Default:: False
Environment Variable:: AIRFLOW__KAFKA_EVENT_PRODUCER__TASK_INSTANCE_EVENTS_ENABLED

task_instance_task_id_allowlist¶

Comma-separated glob patterns. When set, TaskInstance events are only emitted for task_ids matching at least one pattern. Applied in addition to task_instance_dag_id_allowlist — both must pass. Mapped task instances share the same task_id so a single pattern covers all map indices.

Type:: string
Default:: ''
Environment Variable:: AIRFLOW__KAFKA_EVENT_PRODUCER__TASK_INSTANCE_TASK_ID_ALLOWLIST
Example:: load_*,extract_*

task_instance_task_id_denylist¶

Comma-separated glob patterns. TaskInstance events for task_ids matching any pattern are skipped. Deny takes precedence over allow.

Type:: string
Default:: ''
Environment Variable:: AIRFLOW__KAFKA_EVENT_PRODUCER__TASK_INSTANCE_TASK_ID_DENYLIST
Example:: *_cleanup

topic¶

Topic the plugin publishes events to. The topic must already exist on the broker; the plugin will not auto-create it.

Type:: string
Default:: airflow.events
Environment Variable:: AIRFLOW__KAFKA_EVENT_PRODUCER__TOPIC

topic_check_retry_interval¶

How long (in seconds) to wait before retrying a topic check, in case it failed.

Type:: integer
Default:: 60
Environment Variable:: AIRFLOW__KAFKA_EVENT_PRODUCER__TOPIC_CHECK_RETRY_INTERVAL

topic_check_timeout¶

How long (in seconds) each topic existence check is allowed to block waiting for a response from the broker.

Type:: integer
Default:: 10
Environment Variable:: AIRFLOW__KAFKA_EVENT_PRODUCER__TOPIC_CHECK_TIMEOUT

Highlighted configurations ¶

The [kafka_event_producer] section configures the KafkaEventProducerPlugin, which publishes Airflow DagRun and TaskInstance state-change events to a Kafka topic. DagRun and TaskInstance events are separated and enabled by distinct flags. Both event-type flags default to False.

Common use-cases¶

Warning

This is coordination between distinct Airflow deployments, each with its own metadata database. It should not be confused with Airflow’s multi-team feature, which has not been tested with this plugin. Multi-team is a separate, experimental feature still in preview — see Multi-Team for what it provides and its current status.

Activating the plugin¶

To enable event publishing you need to

[kafka_event_producer]
dag_run_events_enabled = True
task_instance_events_enabled = True
kafka_config_id = kafka_events
topic = airflow.events

The connection’s extra JSON accepts the full confluent-kafka client configuration — including SASL/TLS options and callbacks (e.g. error_cb, oauth_cb) given as dotted-path strings. A string-valued callback is only resolved when its full importable path is listed in the callback_allowlist option; the example below requires callback_allowlist = my_company.auth.oauth_cb. This is enforced for security reasons, to prevent malicious callbacks from being executed.

{
    "bootstrap.servers": "broker:9092",
    "security.protocol": "SASL_SSL",
    "sasl.mechanisms": "OAUTHBEARER",
    "oauth_cb": "my_company.auth.oauth_cb"
}

Environment-variable equivalents:

AIRFLOW__KAFKA_EVENT_PRODUCER__DAG_RUN_EVENTS_ENABLED=True
AIRFLOW__KAFKA_EVENT_PRODUCER__TASK_INSTANCE_EVENTS_ENABLED=True
AIRFLOW__KAFKA_EVENT_PRODUCER__KAFKA_CONFIG_ID=kafka_events
AIRFLOW__KAFKA_EVENT_PRODUCER__TOPIC=airflow.events

The two event flags are independent, users can opt-in to get only DagRun event messages or only TaskInstance event messages or both.

The topic must already exist on the broker, it’s not auto-created. On a missing topic, broker connection failure, or any other producer init error, the plugin doesn’t fail, instead it logs a warning and retries the init after topic_check_retry_interval seconds (default 60). Once the topic is created on the broker the plugin will pick it up.

Filtering events¶

DagRun and TaskInstance events are filtered separately. Each filter is a comma-separated list of fnmatch glob patterns; an empty list means “allow all”, and deny takes precedence over allow.

[kafka_event_producer]
dag_run_dag_id_allowlist = sales_*,marketing_*
dag_run_dag_id_denylist = sales_internal_*

task_instance_dag_id_allowlist = sales_*
task_instance_dag_id_denylist =
task_instance_task_id_allowlist = load_*,extract_*
task_instance_task_id_denylist = *_cleanup

TaskInstance events must pass both the dag-id and task-id filters. Mapped task instances share their parent’s task_id, so a single task_id pattern covers every map index.