Enable Kafka coordinator fencing when committing consumer offsets

[twthorn]

Feature Request / Improvement

Description

Add support for coordinator fencing. This can prevent stale ("zombie") coordinators from committing stale consumer offsets.

Implementation

Currently, we generate a new UUID for every new coordinator. This means that the Kafka transaction we use trivially passes since no epoch fencing can be performed on the newly generated ID (e.g., between a stale or an up to date coordinator).

The change is to use a persistent transactional ID for the coordinator. Thus, any leader change of the coordinator group will trigger Kafka to update the epoch of the transaction, and correctly fence stale coordinators. A stale coordinator will identify that it has been fenced, and exit the coordinator thread. We leave the worker task running and defer to Kafka Connect on determining its lifecycle.

See PR for more details

Query engine

None

Willingness to contribute

• I can contribute this improvement/feature independently
• I would be willing to contribute this improvement/feature with guidance from the Iceberg community
• I cannot contribute this improvement/feature at this time

[twthorn]

PR published #18039