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.

[kafka_event_producer]¶

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

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¶

Consume the Kafka events by an external observability or analytics tool and gather info about the state of multiple Airflow instances without polling their metadata DBs.

Based on the state of a DagRun, trigger a downstream external system/pipeline (notifications, alerting, cross-team handoffs) without direct interaction with Airflow.

Coordinate Dags across multiple Airflow instances over a shared Kafka service.

For example, team_A with Airflow instance_A has a deferred task which is triggered when a task from team_B with Airflow instance_B finishes.

Activating the plugin¶

To enable event publishing you need to

enable at least one event-type flag

point the plugin at an Airflow Kafka connection via kafka_config_id (defaults to kafka_default) that carries the broker address and any other confluent-kafka client options on its extras

have a pre-existing kafka topic

[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, which are resolved to callables before the producer is built.

{
    "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.