Kafka Connect: Rework commit-coordinator leader election and harden the coordinator

[kumarpritam863]

What

Reworks how the Iceberg Kafka Connect sink elects its single commit Coordinator and
hardens the coordinator's fencing, recovery, and shutdown paths. No control-topic
wire-format change; exactly-once semantics are preserved.

Leader election — level read, no Admin call

• Before: the coordinator was chosen on every rebalance by calling
  Admin.describeConsumerGroups(connectGroupId) and picking the task that owned the
  globally-lowest (topic, partition) from a transient, mid-rebalance member snapshot.
  This required a DESCRIBE ACL, added an Admin round-trip to the rebalance path, and was
  racy during cooperative rebalancing.
• After: the leader is the task whose assignment() contains partition 0 of the
  lexicographically-smallest topic in its own subscription() — connector-wide,
  identical across tasks, and valid for both a topics list and a topics.regex. No Admin
  call.
• Leadership is read as a level on the task thread. open/close only flag that a
  reconcile is needed; save() reconciles once — start the coordinator if this task owns
  the leader partition, stop it otherwise (both idempotent). Keeping start/stop off the
  rebalance callback avoids blocking RPCs there and stops an eager rebalance from
  needlessly restarting a still-leading coordinator (which would discard in-flight commit
  state).
• The coordinator's commit-readiness partition count now comes from
  consumer.partitionsFor(...) over the subscribed topics instead of a member snapshot.
• Adds a Committer.configure(Catalog, IcebergSinkConfig, SinkTaskContext) lifecycle hook
  (invoked from IcebergSinkTask.start) for one-time setup.

Coordinator hardening

• Zombie fencing via a fixed transactional.id. The coordinator producer id is now
  <connectGroupId>-<connectorName>-coord — identical across a connector's tasks and
  stable across restarts — so a newly elected coordinator's initTransactions()
  epoch-fences a prior (zombie) coordinator's control-plane writes. Worker ids are
  unchanged. This fences the brief two-coordinator overlap the level-read handoff can
  create (the losing task stops on its next save()).
• Fenced ≠ fatal. A coordinator that terminates because it was fenced
  (ProducerFenced / InvalidProducerEpoch / UnknownProducerId, matched across the
  cause chain) is cleared without failing the task; any other termination still fails the
  task.
• Recovery reads from earliest. The -coord consumer group defaults to
  auto.offset.reset=earliest, so a fresh or expired group re-reads uncommitted control
  events instead of skipping to the log end. Replay is idempotent via the snapshot offset
  floor + distinctByKey(location) dedup + the offsets compare-and-swap.
• Idempotency filter in Channel.consumeAvailable. Control-topic records at or below
  the already-consumed per-partition offset are skipped, so a re-delivered or rewound
  record is never re-buffered or re-counted.
• Transient Kafka commit errors are retryable. Errors from commitConsumerOffsets
  (Iceberg CommitFailedException, Kafka CommitFailedException,
  RebalanceInProgressException, RetriableException) are retried rather than failing the
  task — the table commit already succeeded.
• Bounded, interrupt-safe shutdown. stopCoordinator clears state first, then
  bounded-joins the thread; terminate() failures are best-effort and interrupts are
  preserved.
• No producer leak on failed init. KafkaClientFactory.createProducer closes the
  producer if initTransactions() throws.

Compatibility

• No config or control-topic wire-format changes.
• Exactly-once unchanged — anchored on the Iceberg offsets compare-and-swap +
  file-location dedup; the fixed transactional.id adds control-plane fencing on top.

Testing

• Unit tests for the election key (leaderPartition), the retryable-commit classification,
  and coordinator fenced-vs-fatal termination.
• Integration suite passes (13 tests).
• Raised the integration-test commit-wait from 30s to 60s to reduce CI-load flakiness.

[kumarpritam863]

@laskoviymishka @danicafine @bryanck can you please review.

[laskoviymishka]

Thanks for this — dropping the describeConsumerGroups() round-trip at open() and computing leadership locally is a nice simplification, and most of the coordinator hardening (bounded stop-join, offset-skip on replay, treating fencing as non-fatal, closing the producer on a failed init) is the right call.

I'd hold it before merge, though — a couple of these changes read as correctness regressions rather than pure hardening, and I've left the details inline. The two I'd most want to settle:

• The coordinator transactional-id change (format flip + dropped transactionalSuffix) breaks zombie fencing across a rolling upgrade and collapses multi-cluster isolation onto one id, so two coordinators can be live in the overlap window and the loser can still advance control-topic offsets — a later coordinator then skips events.
• A null/empty partitionsFor result silently yields a partition count of 0, which I think lets the first DATA_COMPLETE trigger a full commit and drop the other tasks' still-buffered files.

Plus two more in the new reconcile path: clearing a fenced coordinator doesn't re-arm reconcileNeeded (a leader can silently stop committing until the next rebalance), and per-task subscription() makes leader election ambiguous under topics.regex. And I'd be surgical about the isRetryable broadening — a TimeoutException from a transactional commit is ambiguous to retry.

A few smaller things I left out to keep this focused, worth a look but not blocking: Utils.closeQuietly in Channel.stop() swallows producer-close failures (a dangling open txn goes invisible); the new Committer.configure() default no-op makes an init-ordering contract a custom committer can silently skip; the unused two-arg NotRunningException would let processControlEvents() chain the real cause instead of dropping it; and the core configure()/open()/save() reconcile path (including the fenced-clear branch) isn't covered by a test — the new unit test only exercises the static leaderPartition helper.

Good rework overall, and I'd be happy to take another pass once the fencing/upgrade story and the partition-count path are sorted. wdyt?

[laskoviymishka]

I think this is decently looking to me now.
I would ask @AnatolyPopov to look before merge.

Plus @kumarpritam863 - need to fix the tests