[PR] Fix Kafka consumer not being closed on error in ConsumeFromTopicOperator [airflow]

["TimurRakhmatullin86 (via GitHub)" <gi...@apache.org>]

TimurRakhmatullin86 opened a new pull request, #69641:
URL: https://github.com/apache/airflow/pull/69641

   `ConsumeFromTopicOperator.execute` only called `consumer.close()` on the happy path. If `apply_function` / `apply_function_batch` raised, or `consumer.consume()` / `consumer.commit()` failed, the operator leaked the consumer: the group member stays registered on the broker until `session.timeout.ms` expires (delaying partition rebalance for the rest of the consumer group), and the underlying librdkafka handle keeps its sockets and background threads alive in the worker until GC.
   
   This wraps the processing loop in `try`/`finally` so the consumer is always closed:
   
   - the loop body is unchanged (re-indent only);
   - offset commits stay inside `try`, so nothing is committed for a failed batch;
   - `close()` in `finally` is guarded with a `log.warning`, mirroring the `AwaitMessageTrigger` consumer cleanup from #64612 — `close()` performs a final synchronous offset commit when `enable.auto.commit` is on (the librdkafka default), so it can itself raise and would otherwise replace the original processing error as the task failure reason.
   
   Tests: `test_execute_closes_consumer_when_apply_function_raises` (fails without the fix) and `test_execute_does_not_mask_error_when_close_raises`.
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Fix Kafka consumer not being closed on error in ConsumeFromTopicOperator [airflow]

["potiuk (via GitHub)" <gi...@apache.org>]

potiuk merged PR #69641:
URL: https://github.com/apache/airflow/pull/69641


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org