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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
