GitHub user PrithwishDas101 deleted a comment on the discussion: Make Kafka
AwaitMessageSensor be able to not commit offset
This capability has since been added to `AwaitMessageSensor` through the
`commit_offset` parameter.
You can now disable automatic offset commits by setting `commit_offset=False`:
listen_for_message = AwaitMessageSensor(
task_id="wait_for_kafka_msg",
kafka_config_id="kafka_test_con",
topics=["messages_topic"],
apply_function=None,
commit_offset=False,
)
With this setting, the sensor won't commit the offset automatically, so a
downstream task can handle committing after successful processing. The option
was added in `apache-airflow-providers-apache-kafka` 1.13.0.
One important detail for your specific workflow: if the downstream task needs
to commit the exact message it processed, make sure the sensor passes enough
metadata through XCom. With `apply_function=None`, the trigger returns the
decoded message value, not the message's topic, partition, and offset.
You can use a dotted-path `apply_function` that receives the Kafka message and
returns a serializable dictionary containing those fields along with the
payload. The downstream task can then use that metadata to commit the
appropriate offset, accounting for Kafka's next-offset commit semantics.
Also, ensure that the downstream consumer uses the intended consumer group and
handles retries carefully, so a failed processing task doesn't accidentally
acknowledge a message before it is successfully processed.
References:
- [[AwaitMessageSensor API
documentation](https://airflow.apache.org/docs/apache-airflow-providers-apache-kafka/stable/_api/airflow/providers/apache/kafka/sensors/kafka/index.html)](https://airflow.apache.org/docs/apache-airflow-providers-apache-kafka/stable/_api/airflow/providers/apache/kafka/sensors/kafka/index.html)
- [[AwaitMessageTrigger
implementation](https://airflow.apache.org/docs/apache-airflow-providers-apache-kafka/stable/_modules/airflow/providers/apache/kafka/triggers/await_message.html)](https://airflow.apache.org/docs/apache-airflow-providers-apache-kafka/stable/_modules/airflow/providers/apache/kafka/triggers/await_message.html)
GitHub link:
https://github.com/apache/airflow/discussions/62854#discussioncomment-18946772
----
This is an automatically sent email for [email protected].
To unsubscribe, please send an email to: [email protected]
