Apache Airflow Apache Kafka Provider

Warnings

• breaking The 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.
  Fix: Ensure 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.
• gotcha The 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.
  Fix: Verify `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.
• gotcha Incorrect 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.
  Fix: Carefully 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.
• deprecated The 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.
  Fix: Update your `requirements.txt` to use `apache-airflow-providers-apache-kafka` and adjust import paths if necessary to `airflow.providers.apache.kafka.*`.
• gotcha Apache 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.
  Fix: Use 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.

Install

• pip install apache-airflow-providers-apache-kafka Install the provider

Imports

• ProduceToTopicOperator

  from airflow.providers.apache.kafka.operators.produce import ProduceToTopicOperator

  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

  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

Quickstart

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