Registry /
workflow / apache-airflow-providers-apache-kafka
Install & Compatibility
Where this runs
tested against v1.16.0 · pip install
no network on importno background threads
Install × environment matrix
Each cell = how many times install + import succeeded across repeated harness runs. Partial = flaky.
glibc = Debian/Ubuntu slim · musl = Alpine Linux
muslpy 3.10–3.95 runs
build_error
glibcpy 3.10–3.95 runs
installs and imports cleanly · install 24.9s · import 5.416s · 270MB
270MB installed
● package 270MB
Code
Verified usage
Verified import paths — ran on the pinned version, not inferred.
ProduceToTopicOperator
✓ from airflow.providers.apache.kafka.operators.produce import ProduceToTopicOperator
✗ from airflow.providers.kafka.operators.produce import KafkaProducerOperator
The official package uses `apache.kafka` in the path and `ProduceToTopicOperator` for the class name, not `kafka` or `KafkaProducerOperator` directly.
ConsumeFromTopicOperator
✓ from airflow.providers.apache.kafka.operators.consume import ConsumeFromTopicOperator
✗ from airflow.providers.kafka.operators.consume import KafkaConsumerOperator
The official package uses `apache.kafka` in the path and `ConsumeFromTopicOperator` for the class name.
AwaitKafkaMessageOperator
✓ from airflow.providers.apache.kafka.operators.await_message import AwaitKafkaMessageOperator
KafkaProducerHook
✓ from airflow.providers.apache.kafka.hooks.producer import KafkaProducerHook
KafkaConsumerHook
✓ from airflow.providers.apache.kafka.hooks.consumer import KafkaConsumerHook
KafkaAdminClientHook
✓ from airflow.providers.apache.kafka.hooks.admin_client import KafkaAdminClientHook
AwaitMessageTrigger
✓ from airflow.providers.apache.kafka.triggers.await_message import AwaitMessageTrigger
This quickstart demonstrates a simple Airflow DAG that uses the Kafka provider to produce and then consume messages from a Kafka topic. Before running, ensure you have a Kafka connection configured in the Airflow UI (Admin -> Connections) named `kafka_default`. The 'Extras' field should contain `{"bootstrap.servers": "broker:9092"}` or the appropriate Kafka broker address. The example assumes a local Kafka instance accessible via `broker:9092`.
from __future__ import annotations
import pendulum
from airflow.models.dag import DAG
from airflow.providers.apache.kafka.operators.produce import ProduceToTopicOperator
from airflow.providers.apache.kafka.operators.consume import ConsumeFromTopicOperator
with DAG(
dag_id="kafka_producer_consumer_example",
start_date=pendulum.datetime(2023, 1, 1, tz="UTC"),
catchup=False,
schedule=None,
tags=["kafka", "example"],
) as dag:
# Define a Kafka connection in Airflow UI named 'kafka_default'
# e.g., Host: broker:9092, Port: <empty>, Extras: {"bootstrap.servers": "broker:9092"}
produce_messages = ProduceToTopicOperator(
task_id="produce_test_messages",
topic="test_topic",
kafka_conn_id="kafka_default",
producer_config={'bootstrap.servers': 'broker:9092'},
messages=[{'key': f'key-{i}', 'value': f'value-{i}'} for i in range(5)],
)
consume_messages = ConsumeFromTopicOperator(
task_id="consume_test_messages",
topic="test_topic",
kafka_conn_id="kafka_default",
consumer_config={'bootstrap.servers': 'broker:9092', 'group.id': 'airflow-consumer-group', 'auto.offset.reset': 'earliest'},
apply_function='lambda msg: print(f"Consumed: {msg.key}, {msg.value}")', # Simple print function
commit_messages_every_n_messages=1, # Commit after each message for demonstration
max_messages=5, # Consume up to 5 messages
poll_timeout=10, # Max time to wait for messages
)
produce_messages >> consume_messages
airflow --version
Debug
Known issues
breakingThe minimum supported Apache Airflow version for the `apache-airflow-providers-apache-kafka` provider is `2.11.0` for provider version `1.13.1`. Earlier provider versions (e.g., 1.7.0, 1.9.0, 1.11.0) required older minimum Airflow versions. Upgrading the provider might necessitate an upgrade of your Airflow core installation.fixEnsure your Apache Airflow core version meets or exceeds the minimum requirement for the installed provider version. Refer to the provider's changelog or documentation for specific compatibility matrices.
affects: All provider versions, specifically 1.13.1 and later.
gotchaThe provider relies on the `confluent-kafka` Python package. Issues, especially related to the underlying `librdkafka` C library (e.g., when Kerberos authentication is involved), can cause connection failures.fixVerify `confluent-kafka` is installed correctly (`pip install confluent-kafka`). For complex scenarios like Kerberos, you might need to build `librdkafka` from source and install `confluent-kafka` with the `--no-binary` flag. Ensure all necessary system-level dependencies for `librdkafka` (e.g., `libsasl2-dev`, `krb5-user`) are present.
affects: All versions
gotchaIncorrect or incomplete Kafka connection configurations in the Airflow UI (or `kafka_conn_id` in operators) are a common source of errors, leading to connection failures, missing messages, or unexpected behavior. Key parameters like `bootstrap.servers` and `group.id` (for consumers) are critical.fixCarefully configure your Kafka connection in the Airflow UI. For `Apache Kafka` connection type, specify `bootstrap.servers` in the 'Extras' JSON field. For consumers, ensure a `group.id` is provided and is unique across consumer groups. Consult `librdkafka` documentation for a full list of connection parameters.
affects: All versions
deprecatedThe original `astronomer/airflow-provider-kafka` GitHub repository and its associated PyPI package have been discontinued since March 2023. Users should migrate to the official `apache-airflow-providers-apache-kafka` package.fixUpdate your `requirements.txt` to use `apache-airflow-providers-apache-kafka` and adjust import paths if necessary to `airflow.providers.apache.kafka.*`.
affects: Users of `astronomer-airflow-providers-kafka`
gotchaApache Airflow is designed for workflow orchestration and batch processing, not for real-time streaming or low-latency operations. While the Kafka provider allows interaction with Kafka, using Airflow itself for managing continuous streaming processes is an anti-pattern and can lead to performance and operational issues.fixUse Airflow to orchestrate tasks that *interact* with Kafka (e.g., trigger a data load after a Kafka topic reaches a certain state, or publish results to Kafka). Avoid using Airflow as a streaming engine; dedicated streaming platforms are better suited for low-latency, high-throughput stream processing.
affects: All versions
Errors
Common errors & fixes
ModuleNotFoundError: No module named 'airflow.providers.apache.kafka'
The 'apache-airflow-providers-apache-kafka' package is not installed.
fixInstall the package using pip: 'pip install apache-airflow-providers-apache-kafka'.
ImportError: cannot import name 'KafkaOperator' from 'airflow.providers.apache.kafka.operators'
The 'KafkaOperator' class is not available in the specified module.
fixEnsure you are importing the correct class from the correct module, and that your 'apache-airflow-providers-apache-kafka' package is up to date.
AttributeError: module 'airflow.providers.apache.kafka.hooks' has no attribute 'KafkaHook'
The 'KafkaHook' attribute does not exist in the 'airflow.providers.apache.kafka.hooks' module.
fixVerify the correct import path and ensure that the 'apache-airflow-providers-apache-kafka' package is properly installed and up to date.
ModuleNotFoundError: No module named 'confluent_kafka'
The 'confluent-kafka' package, a dependency of 'apache-airflow-providers-apache-kafka', is not installed.
fixInstall the 'confluent-kafka' package using pip: 'pip install confluent-kafka'.
TypeError: 'NoneType' object is not iterable
A function is returning 'None' when an iterable is expected, possibly due to a misconfigured Kafka connection.
fixCheck the Kafka connection configuration in Airflow and ensure all required parameters are correctly set.
Upgrade
Version history
1.16.0latest on PyPI · released Aug 23, 2026
Audit
Dependencies
apache-airflowrequiredCore Airflow installation is required. Version 1.13.1 of the provider requires Apache Airflow >=2.11.0.
confluent-kafkarequiredThe provider uses the confluent-kafka-python library for interacting with Kafka. Specific versions are required based on Python version (e.g., >=2.6.0 for Python < 3.14, >=2.13.2 for Python >= 3.14).
apache-airflow-providers-common-compatrequiredA common compatibility provider package required by Apache Airflow providers, version >=1.12.0.
asgirefrequiredASGI server and client utilities, version requirements depend on Python version.