Add KafkaSharedStreamProducer and KafkaSharedStreamTrigger

[FrankYang0529]

Adds KafkaSharedStreamProducer and KafkaSharedStreamTrigger to the Apache Kafka provider, backing the producer-side ack channel for shared-stream triggers (#67523). Many event-driven triggers / asset watchers on the same Kafka topics can now share a single consumer in the trigger, with offset commits gated on trigger-event persistence (at-least-once).

Verification

1. Prepare files

mkdir -p files/dags && cat > files/dags/demo_kafka_shared_stream.py <<'EOF'
from __future__ import annotations

from airflow.providers.apache.kafka.triggers.shared_stream import KafkaSharedStreamTrigger
from airflow.providers.standard.operators.empty import EmptyOperator
from airflow.sdk import DAG, Asset, AssetWatcher

TOPIC = "shared-stream-demo"
CONN_ID = "kafka_shared_demo"

orders_trigger = KafkaSharedStreamTrigger(topics=[TOPIC], kafka_config_id=CONN_ID, poll_timeout=1.0)
audit_trigger = KafkaSharedStreamTrigger(topics=[TOPIC], kafka_config_id=CONN_ID, poll_timeout=2.0)

orders_asset = Asset(
    "kafka_shared_orders",
    watchers=[AssetWatcher(name="orders_watcher", trigger=orders_trigger)],
)
audit_asset = Asset(
    "kafka_shared_audit",
    watchers=[AssetWatcher(name="audit_watcher", trigger=audit_trigger)],
)

with DAG(dag_id="kafka_shared_stream_orders", schedule=[orders_asset], catchup=False):
    EmptyOperator(task_id="handle_order")

with DAG(dag_id="kafka_shared_stream_audit", schedule=[audit_asset], catchup=False):
    EmptyOperator(task_id="handle_audit")
EOF

These environment variables are needed to avoid a but. #72138 will fix it.

mkdir -p files/airflow-breeze-config && cat > files/airflow-breeze-config/environment_variables.env <<'EOF'
export AIRFLOW__KAFKA_EVENT_PRODUCER__DAG_RUN_EVENTS_ENABLED=False
export AIRFLOW__KAFKA_EVENT_PRODUCER__TASK_INSTANCE_EVENTS_ENABLED=False
EOF

2. Start Airflow

breeze start-airflow --integration kafka --backend postgres

3. Create a Topic (Terminal B)

docker exec broker kafka-topics --bootstrap-server localhost:9092 --create --topic shared-stream-demo --partitions 1 --replication-factor 1

4. Enter the container (Terminal B)

breeze exec

5. Create a Kafka connection (Terminal B)

airflow connections add kafka_shared_demo --conn-type kafka --conn-extra '{"bootstrap.servers": "broker:29092", "group.id": "shared-stream-demo-group", "enable.auto.commit": false, "auto.offset.reset": "beginning"}'

6. Unpause dags and check there are two triggers(Terminal B)

airflow dags unpause kafka_shared_stream_orders
airflow dags unpause kafka_shared_stream_audit

python -c "
from airflow.models.trigger import Trigger
from airflow.utils.session import create_session
with create_session() as s:
    for t in s.query(Trigger).all():
        print(t.id, t.classpath.rsplit('.', 1)[-1])
"

Expected output:

1 KafkaSharedStreamTrigger
2 KafkaSharedStreamTrigger

7. Check only one shared consumer (Terminal B)

docker exec broker kafka-consumer-groups --bootstrap-server localhost:9092 --describe --group shared-stream-demo-group --members

8. Producer message (Terminal B)

printf 'order-1\norder-2\norder-3\n' | docker exec -i broker kafka-console-producer --bootstrap-server localhost:9092 --topic shared-stream-demo

9. Both dags have succeeded run (Terminal B)

airflow dags list-runs kafka_shared_stream_orders
airflow dags list-runs kafka_shared_stream_audit

10. Check Kafka offset is committed (Terminal B)

docker exec broker kafka-consumer-groups --bootstrap-server localhost:9092 --describe --group shared-stream-demo-group

related: #67523

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

• Yes (Claude Code with Opus 4.8)

---

• 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.

[jason810496]

Nice! LGTM overall.
It would be nice to manually test with breeze start-airflow --integration kafka setup when you have a moment, thanks.

[Lee-W]

will need some time to look it through, but found one nit

[github-actions]

This pull request has been automatically marked as stale because it has not had recent activity. It will be closed in 5 days if no further activity occurs. Thank you for your contributions.

[jason810496]

It would be nice to manually test with breeze start-airflow --integration kafka setup when you have a moment, thanks.

Looking forward for the real integration setup to exercise this feature when you have a moment, thanks.

[FrankYang0529]

@jason810496 I update real integration setup to PR description and verify it can work. Thank you.