KafkaMessageQueueTrigger resulting AssetEvent diverges from _consumer_offsets

[spencerhuang]

Description

Using Airflow 3.1.8

Hi all,

I've been experimenting with Kafka Message Queue Trigger. As in CDC Debezium -> Kafka -> AssetWatcher + Kafka Message Queue Trigger -> AssetEvent -> DAG

It's somewhat disheartening. One day in, I noticed that AssetEvent table and Kafka's _consumer_offsets are starting to diverge. Stale AssetEvent rows in Airflow's Postgres Metastore were separate from Kafka offsets. Resetting Kafka consumer offsets had no effect on these already-queued events. Each stale event consumed a DAG run, blocking the next actual event from being processed within the timeout. It's likely caused by a network partition.

Is this a low probability scenario, or is this a practical concern? Most business would require Strict Exactly-Once. My understanding is that this is not a matter of If, but how to handle the fallout when it happens. Is there a way to improve upon this, any ideas?

Use case/motivation

A way to detect drift or divergence and a contingency/approach to re-align with Kafka.

Related issues

No response

Are you willing to submit a PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[boring-cyborg]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.

[jroachgolf84]

Stale AssetEvent rows in Airflow's Postgres Metastore were separate from Kafka offsets.

Can you expand on this a little bit more here? What do you mean by "separate"?

I'll preface this by saying that I am NOT a Kafka expert by any means. But for all MessageQueueTriggers (including Kafka), the logic is quite simple. When a message lands in the message queue an AssetEvent is created, triggering a downstream DAG (or DAGs). What's really being executed is this code (https://github.com/apache/airflow/blob/main/providers/apache/kafka/src/airflow/providers/apache/kafka/triggers/await_message.py#L106-L143), just checking for a simple event.

[spencerhuang]

Stale AssetEvent rows in Airflow's Postgres Metastore were separate from Kafka offsets.

Can you expand on this a little bit more here? What do you mean by "separate"?

I'll preface this by saying that I am NOT a Kafka expert by any means. But for all MessageQueueTriggers (including Kafka), the logic is quite simple. When a message lands in the message queue an AssetEvent is created, triggering a downstream DAG (or DAGs). What's really being executed is this code (https://github.com/apache/airflow/blob/main/providers/apache/kafka/src/airflow/providers/apache/kafka/triggers/await_message.py#L106-L143), just checking for a simple event.

@jroachgolf84 Thank you for the quick response. Let me provide some contexts on how I use Kafka in conjunction with Airflow.

In Kafka, a Poison Pill is a specific message that consistently fails to be processed, often due to a corrupted message, causing the consumer to get "stuck" because it cannot move past the failing offset. To prevent this single message from stopping the entire data pipeline, we use a Dead Letter Queue (DLQ). A DLQ is a separate, dedicated Kafka topic where the consumer automatically reroutes these problematic messages after a few failed attempts. This allows the main pipeline to continue running smoothly while isolating the "poison" data for manual inspection and fixing later.

To address this classic, common Kafka issue, here is the use-case:

# Event-driven DAG triggered by the cdc_events Asset:
with DAG(
    dag_id="cdc_processor",
    schedule=[cdc_asset],  # Triggered by AssetEvents
    ...
) as dag:
    result = process_and_trigger()   # @task (retries=3, exp backoff)
    handle_dlq_task = handle_dlq()   # @task (trigger_rule=ONE_FAILED)
    result >> handle_dlq_task

process_and_trigger: Reads event payload from context["triggering_asset_events"], queries the data from a business DB, builds DAG run conf, starts the on-demand DAG via Airflow REST API, and maybe emits audit event.

handle_dlq: Fires when process_and_trigger fails after 3 retries. Persists the failed message to the dead_letter_messages table and Kafka DLQ topic.

Now, the diverge, or separate/separation

A malformed Kafka message causes a performance degradation or crash in the Triggerer before a DLQ write can occur. To restore service, an on-call/support will naturally perform a Kafka offset reset (current + 1) to skip the bad record. However, because the AssetWatcher may have already partially emitted AssetEvents into Postgres before the stall/crash, the offset reset is insufficient. Airflow will still "phantom-process" those ghost events from its own database, even though the operator explicitly signaled the system to skip that data. We now have two Separate competing "Sources of Truth" (Kafka offsets vs. Airflow Asset state) that are fundamentally out of sync.

This architectural separation would create problems during recovery also. If an operator resets Kafka to Latest to shed load during a lag spike or a slowdown in Airflow, they expect the system to stop processing old data. Instead, Airflow continues to churn through the backlog of pending AssetEvents stored in the metadata DB. Without a unified state, we lose the ability to perform a Point-in-Time Recovery. Resetting one does not reset the other, making a clean "replay" of Kafka events difficult to coordinate. By decoupling the ingestion state from the orchestration state, we have traded operational simplicity for a system where the left hand (Kafka) no longer knows what the right hand (Airflow) is doing. I hope this clarifies why the current separation is a significant concern for production-grade, event-driven pipelines.

[jason810496]

Hi @spencerhuang,

Thank you for the detailed description.

We now have two Separate competing “Sources of Truth” (Kafka offsets vs. Airflow Asset state) that are fundamentally out of sync.⁠

After testing this myself, I would suggest implementing a custom trigger to handle the processing and DLQ logic atomically if you really need strict guarantees of messaging delivery semantics. Since the Airflow Triggerer handles fired triggers in batches, it commits the offset before creating the ‎AssetEvent. This means that if you set ‎commit_offset=True on the current community ‎AwaitMessageTrigger, there is still a chance that a corresponding ‎AssetEvent will not be produced even though the offset has already been committed.

After implementing the custom trigger that handles the processing and DLQ logic atomically, you can then implement ‎BaseMessageQueueProvider to register it for ‎MessageQueueTriggers:

airflow/providers/apache/kafka/src/airflow/providers/apache/kafka/queues/kafka.py

Line 34 in 1790ec4

class KafkaMessageQueueProvider(BaseMessageQueueProvider):

[jason810496]

Additionally, based on your use case, here is a working example of handling a poison pill message automatically. We can then have another event‑driven DAG to process messages from the ‎DLQ topic.

However, please note that, as I mentioned earlier, there is still a chance that we commit the offset before we actually create the corresponding ‎AssetEvent. This is a known limitation, and the only way to address it is to have a custom ‎Trigger that handles processing messages and routing malformed messages to the DLQ in an atomic function before ‎yield TriggerEvent.

from __future__ import annotations

import json
from typing import TYPE_CHECKING

import pendulum

from airflow.providers.apache.kafka.operators.produce import ProduceToTopicOperator
from airflow.providers.common.messaging.triggers.msg_queue import MessageQueueTrigger
from airflow.sdk import Asset, AssetWatcher, dag, get_current_context, task

if TYPE_CHECKING:
    from airflow.sdk.execution_time.context import Context, TriggeringAssetEventsAccessor

KAFKA_CONFIG_ID = "issue_64205_kafka_config"
TOPICS = ["fizz_buzz"]
DLQ_TOPIC = "dlq"
RETRY_COUNT = 3
"""Airflow Kafka connection
export AIRFLOW_CONN_ISSUE_64205_KAFKA_CONFIG='{
  "conn_type": "general",
  "extra": {
    "bootstrap.servers": "broker:29092",
    "group.id": "issue_64205_group",
    "security.protocol": "PLAINTEXT",
    "enable.auto.commit": false,
    "auto.offset.reset": "latest"
  }
}'
"""

SAMPLE_ORDERS = [
    {"order_id": "ORD-1001", "customer": "alice", "item": "widget", "quantity": 2, "price": 9.99},
    {"order_id": "ORD-1002", "customer": "bob", "item": "gadget", "quantity": 1, "price": 24.50},
    {"order_id": "ORD-1003", "customer": "carol", "item": "widget", "quantity": 5, "price": 9.99},
    {"order_id": "ORD-1004", "customer": "dave", "item": "doohickey", "quantity": 3, "price": 14.75},
    {"order_id": "ORD-1005", "customer": "eve", "item": "thingamajig", "quantity": 1, "price": 39.00},
    {"order_id": "ORD-1006", "customer": "frank", "item": "widget", "quantity": 10, "price": 9.99},
    {"order_id": "ORD-1007", "customer": "grace", "item": "gadget", "quantity": 2, "price": 24.50},
    {"order_id": "ORD-1008", "customer": "heidi", "item": "doohickey", "quantity": 1, "price": 14.75},
]


def producer_function():
    for order in SAMPLE_ORDERS:
        yield (json.dumps(order["order_id"]), json.dumps(order))
    # produce a malformed message to demonstrate error handling
    yield ("malformed_message", "malformed_message")


def process_one_message(message: str):
    order = json.loads(message)
    total = order["quantity"] * order["price"]
    print(f"Order {order['order_id']}: {order['quantity']}x {order['item']} = ${total:.2f}")
    return order

def handle_dlq():
    context: Context = get_current_context()
    triggering_asset_events: TriggeringAssetEventsAccessor = context["triggering_asset_events"]
    for event in triggering_asset_events[kafka_cdc_asset]:
        print(f"Handling failed message from event: {event}")
        value = json.dumps({
            "asset": event.asset.model_dump(mode="json"),
            "extra": event.extra,
        })
        yield (json.dumps(event.asset.uri), value)


# Airflow 3 example
# Define a trigger that listens to an external message queue (Apache Kafka in this case)
trigger = MessageQueueTrigger(
    scheme="kafka",
    # the rest of the parameters are used by the trigger
    kafka_config_id=KAFKA_CONFIG_ID,
    topics=TOPICS,
    poll_interval=1,
    poll_timeout=1,
    commit_offset=True,
)

# Define an asset that watches for messages on the queue
kafka_cdc_asset = Asset("kafka_cdc_asset", watchers=[AssetWatcher(name="kafka_cdc", trigger=trigger)])


@dag(
    schedule=[kafka_cdc_asset],
    tags=["event-driven"],
)
def issue_64205_consumer_patched():

    @task(retries=RETRY_COUNT, retry_delay=1)
    def process_message(**context) -> bool:
        # Extract the triggering asset events from the context
        triggering_asset_events: TriggeringAssetEventsAccessor = context["triggering_asset_events"]
        for event in triggering_asset_events[kafka_cdc_asset]:
            # Get the message from the TriggerEvent payload
            print(f"Asset event: {event}")
            process_one_message(event.extra["payload"])
        return True

    @task.short_circuit(trigger_rule="all_done")
    def should_handle_dlq(**context) -> bool:
        """Skip DLQ handling if processing succeeded."""
        # If process_message succeeded, it pushed True to XCom.
        # If it failed (exception after retries), no XCom was pushed -> None.
        upstream_result = context["ti"].xcom_pull(task_ids="process_message")
        return upstream_result is None

    result = process_message()
    dlq_check = should_handle_dlq()
    handle_dlq_task = ProduceToTopicOperator(
        kafka_config_id=KAFKA_CONFIG_ID,
        task_id="handle_dlq",
        topic=DLQ_TOPIC,
        producer_function=handle_dlq,
    )

    result >> dlq_check >> handle_dlq_task
issue_64205_consumer_patched()

@dag(
    description="Load Data to fizz_buzz topic",
    start_date=pendulum.datetime(2022, 11, 1),
    schedule=None,
    catchup=False,
    tags=["event-driven"],
)
def issue_64205_producer_patched():
    ProduceToTopicOperator(
        kafka_config_id=KAFKA_CONFIG_ID,
        task_id="produce_to_topic",
        topic=TOPICS[0],
        producer_function=producer_function,
    )
issue_64205_producer_patched()

[jason810496]

It's more like Airflow internal restriction and Dag authoring pattern instead of actual issue, I will close as completed for now and please feel free to reopen the issue if needed. Thanks.