kafka connect: use a stable coordinator transactional id to fence stale coordinators

[twthorn]

Description

Fixes #18038

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. We name this using the connect group ID which is unique to that connector. Then any leader change of that 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.

Note: we introduced a in-memory monotonic increasing offset guard to make offsets monotonically increasing (values that are non-increasing are not committed). This was a tightly scoped fix to reduce the risk of a coordinator committing old offsets (which can cause data loss). However, it is theoretically possible that offsets could still be updated if an older coordinator wrote a higher offset than it last stored, but is less than another coordinator's written offset. The way to solve this and the canonical one in Kafka is with producer fencing. So we add the offset commit in its own transaction, and defer to kafka for epoch fencing.

Note: we also intentionally make the commit consumer offsets its separate transaction. It is outside the scope of this PR to refactor the offset transaction semantics (eg grouping the control topic offset write with the consumer offset commit). This delivers the minimal set of changes to have fenced consumer offset commits.

Testing

Used an integration test to reproduce this failure scenario first (ie a stale coordinator with an increasing offset but is older than another offset of a newer coordinator). Also added unit tests as well.

Related Work

In-memory monotonic offset guard:

• kafka connect: data loss with default configs due to persistent zombie coordinators #17551
• kafka connect: coordinator only commits offsets when greater than existing offsets #17552

Larger scoped coordinator refactor (PR is stale, has merge conflicts):

• Kafka Connect: Harden commit coordinator (fencing, recovery, dedup, retryable commits) #17376

Related preventative setting offset reset strategy:

• Kafka Connect: Read the control topic from earliest in the coordinator #18006

[twthorn]

@laskoviymishka could you review this when you get a free moment? Thanks!