Make Kafka AwaitMessageSensor be able to not commit offset

[filippzorin]

Hi everyone, I love using Airflow and I've been trying to work with Kafka AwaitMessageSensor (from apache-airflow-providers-apache-kafka) to trigger task when message is produced to particular topic. In my case I'd like to not commit offset when sensor polls messages, I'm going to commit offset when message is processed in next task.
DAG simplified example:

import datetime

from airflow import DAG
from airflow.providers.apache.kafka.sensors.kafka import AwaitMessageSensor
from airflow.operators.bash import BashOperator

dag = DAG(
    dag_id='kafka-await-test',
    start_date=datetime.datetime(2026, 1, 1),
    schedule="@hourly",
)

listen_for_message = AwaitMessageSensor(
    task_id="wait_for_kafka_msg",
    kafka_config_id="kafka_test_con",
    topics=["messages_topic"],
    apply_function=None
)
run_job = BashOperator(
    task_id="run_after_loop",
    bash_command="echo 'message processing and commit offset is here'",
    dag = dag
)
listen_for_message  >> run_job

What do you think about adding option to AwaitMessageSensor to manage commit behavior while sensor polls messages?