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'd hold on this txn-id change — it's the biggest risk in the PR for me.

The format flips from transactionalPrefix + name + transactionalSuffix to connectGroupId + "-" + connectorName + "-coord", so an old coordinator and a new one carry different transactional ids across a rolling upgrade. initTransactions() only fences the same id, so for the overlap window both coordinators are live — Iceberg CAS lets one win per table, but the loser can still advance the control-topic consumer-group offsets, and a later coordinator replaying from those offsets skips events.

It also drops transactionalSuffix() (the worker path keeps it). Operators set that suffix to isolate producers across multiple Connect clusters, so every cluster's coordinator now collapses onto the same id and they fence each other.

I'd keep the suffix on both paths and gate the new format behind config so upgrades have a migration path. (Minor while we're here: connectGroupId already defaults to connect-<connectorName>, so this renders as connect-myconn-myconn-coord — the name lands twice.) wdyt?

[kumarpritam863]

a new one carry different transactional ids across a rolling upgrade. -> This will be the case only during the upgrade and after that it will be stable.
It also drops transactionalSuffix() (the worker path keeps it). Operators set that suffix to isolate producers across multiple Connect clusters, so every cluster's coordinator now collapses onto the same id and they fence each other. -> This still holds as connectGroupId is same as the consumerGroupId and per kafka cluster we can have only one connect cluster with a consumer group id and if two connectors share the consumer group id then they are effectively in the same group.

I'd keep the suffix on both paths and gate the new format behind config so upgrades have a migration path. (Minor while we're here: connectGroupId already defaults to connect-<connectorName>, so this renders as connect-myconn-myconn-coord — the name lands twice.) wdyt? -> yeah this is true when we do not give consumer group id as part of connector config, but I will change this.

[laskoviymishka]

I think there's a data-loss path here when topic metadata isn't cached yet.

partitionsFor(topic) returns null while the consumer hasn't fetched metadata for a topic (broker still loading, topic just created, first assignment after a cold start), and we silently skip it — so topicPartitionCount can come out 0 and the Coordinator is built expecting 0 partitions. With nothing to wait for, the first DATA_COMPLETE looks commit-ready and we run a full commit while the other tasks' files are still buffered, and those get dropped on clearResponses.

The old code summed partitions from the group members, which was authoritative. I'd either restore an authoritative count or treat count == 0 as unknown and never take the full-commit path in that state — a timeout-based partial commit is fine there, dropping files isn't. Could we guard that case?

[kumarpritam863]

broker still loading, topic -> isn't this thew case in the current scenario as well. Also connect gurantees that a save will only be called after the rebalance has completed. So I am expecting that it should not return null.``

The old code summed partitions from the group members, which was authoritative. I'd either restore an authoritative count or treat count == 0 -> even in this case that topic will not be assigned to anyone so I think it will be the same.

[laskoviymishka]

When we clear a fenced coordinator here, we don't re-arm reconcileNeeded, so I think we can end up with no coordinator at all.

If the fence happens after the last reconcileLeadership() (so reconcileNeeded is already false), we null out coordinatorThread and nothing schedules another reconcile — no future save() re-evaluates leadership until the next rebalance. This task is still the leader by assignment, but it silently stops producing commits.

Setting reconcileNeeded = true right after stopCoordinator() in this branch would let the next save() restart it. wdyt?

[kumarpritam863]

Yeah we can do that but this is fenced means there was another coordinator elected somewhere, means it got the leaderPartition so in case we stopCoordinator on this task and that task then looses that leaderPartition, it will be ultimately assigned to some task which will set reconcile needed in the open/close and evaluate on next save.

[laskoviymishka]

This election looks deterministic for a static topics list but I think it breaks for topics.regex.

subscription() returns only the topics this task instance currently resolved/was assigned, not the full cluster set. With a static list every task sees the same set, so leaderPartition agrees. With a regex, task A might see [event-aaa, event-bbb] and task B [event-ccc, ...], so they compute different local-min topics and each can elect itself → two coordinators overlapping, or zero if nobody holds partition 0 of their local min.

Election needs a global view of the subscribed topics for the regex case. Could we resolve it from cluster state, or explicitly reject/document regex here for now?

[kumarpritam863]

With a static list every task sees the same set, so leaderPartition agrees. With a regex, task A might see [event-aaa, event-bbb] and task B [event-ccc, ...], -> I need to check this as as far as I am aware the sunscription holds the entire list of topics even in the case of regex.

Subscribe to all topics matching specified pattern to get dynamically assigned partitions. The pattern matching will be done periodically against all topics existing at the time of check. This can be controlled through the metadata.max.age.ms configuration.

while there can be brief window at the time of create/delete but that is the case in the current scenario as well. WDYT?

[laskoviymishka]

Broadening retry to the whole RetriableException hierarchy is where I'd want to be surgical.

Iceberg's CommitFailedException, the Kafka consumer CommitFailedException, and RebalanceInProgressException are all safe to retry. But a TimeoutException out of producer.commitTransaction() is ambiguous — per the Kafka docs the broker may already have committed — so retrying by starting a fresh transaction and committing again risks a double write. Config-type RetriableExceptions like NotEnoughReplicas* will also retry to the threshold while logging "will retry", which just misleads operators.

I'd enumerate the exceptions we actually treat as transient for a transactional commit rather than catch the superclass. Which ones did you intend to be retryable here?

[kumarpritam863]

Yeah I think CommitFailedException and RebalanceInProgressException should be fine along with Iceberg's CommitFailedException