Fix Kafka consumer not being closed on error in ConsumeFromTopicOperator

[TimurRakhmatullin86]

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 Add cleanup to Kafka AwaitMessageTrigger for consumer management #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.

[potiuk]

Real leak with blast radius beyond the failing task: with close() only on the happy path, any exception from apply_function, consume() or commit() left the group member registered on the broker until session.timeout.ms expired, which delays partition rebalance for every other consumer in the group — not just the task that failed.

I checked the "re-indent only" claim rather than taking it on trust: diffing added against removed lines ignoring whitespace, the only genuinely new lines are try:, finally:, except Exception: and the warning log, with nothing removed that lacks a re-indented counterpart. That is what makes a 154-line diff quick to be confident about.

Two details this gets right that are easy to miss:

Keeping the commits inside the try means a failed batch still commits nothing, so the leak is fixed without quietly shifting offset semantics.

Wrapping close() in its own try/except stops a failing close from replacing the original exception. test_execute_does_not_mask_error_when_close_raises pins exactly that — close.side_effect raises, and the test still asserts the original ValueError("boom") surfaces. Without the inner guard that test fails with the close error instead, which is the failure mode people usually discover in production rather than review.

Acquiring the consumer outside the try is also right: nothing to close if get_consumer() itself fails, and no unbound name for the finally to trip over.

One thing worth noting rather than fixing here: this cleanup idiom, down to the log wording, now exists in both this operator and AwaitMessageTrigger.cleanup() with no shared helper, so the two can drift.

---

Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting