Kafka Connect: Rework commit-coordinator election (drop Admin describeConsumerGroups) and harden coordinator fencing, recovery, and shutdown.

[kumarpritam863]

What

Reworks how the Iceberg Kafka Connect sink elects its single commit Coordinator and hardens the coordinator's fencing, recovery, and shutdown paths.

Election — no Admin API call

• Before: the coordinator was chosen by calling Admin.describeConsumerGroups(connectGroupId) on every rebalance and picking the task that owned the globally-lowest (topic, partition) from a transient, mid-rebalance group 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 that owns partition 0 of the lexicographically-smallest topic in its own consumer subscription() — connector-wide, identical across tasks, and
  valid for both a literal topics list and a topics.regex. No Admin call. Leadership is evaluated in the open/close rebalance callbacks: a task that gains the leader partition
  starts the coordinator; a task that loses it stops it.
• The coordinator's commit-readiness partition count is derived from the consumer's cached partitionsFor(...) over the subscribed topics, instead of a one-time member-assignment
  snapshot.

Coordinator hardening

• Zombie fencing via a fixed transactional.id. The coordinator producer id is now <prefix>coordinator-<connectorName> — identical across a connector's tasks and stable across
  restarts/reconfigurations — so a newly elected coordinator's initTransactions() epoch-bumps and fences a prior (zombie) coordinator's control-plane writes. Worker ids stay per-task
  (<prefix><taskId>-worker-<suffix>) so sibling workers never fence each other.
• Fenced ≠ fatal. A coordinator that terminates because it was fenced (ProducerFencedException / InvalidProducerEpochException / UnknownProducerIdException, matched across the
  cause chain) is cleared without failing the task (Connect does not auto-restart FAILED tasks); any other coordinator termination still fails the task loudly.
• Recovery reads from earliest. The coordinator's -coord consumer group now defaults to auto.offset.reset=earliest, so a fresh or expired group re-reads uncommitted control
  events instead of skipping to the log end and dropping their data. Replay is idempotent via the snapshot offset floor + distinctByKey(location) dedup + the offsets compare-and-swap.
• Rewind-safe control consumption. Channel is now its own ConsumerRebalanceListener; on (re)assignment the coordinator seeks each control-topic partition forward to the offset it
  has already consumed, so an eager rebalance that rewinds the fetch position does not re-read buffered events. The tracked control offset is merged with Long::max, so it never
  regresses.
• Transient Kafka commit errors are retryable. A consumer-offset commit hiccup from commitConsumerOffsets() (Kafka CommitFailedException / RebalanceInProgressException /
  RetriableException, e.g. the -coord consumer briefly evicted when a catalog commit exceeds max.poll.interval.ms) no longer fails the task — the table commit already succeeded and
  the watermark re-advances next cycle; a persistent failure still terminates after commitMaxConsecutiveFailures.
• Bounded, interrupt-safe shutdown. stopCoordinator clears state first, then signals termination and waits (bounded) for the thread to release its producer/consumer/admin before
  returning; a terminate() failure is best-effort (logged, not fatal) and interrupts are preserved (Coordinator.terminate() restores the interrupt flag).
• No producer leak on failed init. KafkaClientFactory.createProducer closes the producer if initTransactions() throws.

API

• Adds a default Committer.configure(Catalog, IcebergSinkConfig, SinkTaskContext) lifecycle hook (invoked from IcebergSinkTask.start) for one-time setup. The deprecated
  start/stop methods are unchanged.

Compatibility

• No config or control-topic wire-format changes.
• Exactly-once semantics unchanged — still anchored on the Iceberg offsets compare-and-swap + file-location dedup; the fixed transactional.id adds control-plane fencing on top.
• Migrate per-connector with a single coordinated restart. A brief overlap of an old per-task-id coordinator and a new fixed-id coordinator is not mutually fenced but is
  offsets-CAS-safe.

Testing

• Unit tests for: the election key (lowest-topic partition 0), open/close start/stop of the coordinator, fenced-vs-fatal coordinator termination (including wrapped,
  InvalidProducerEpoch, and UnknownProducerId causes), the retryable-commit classification, seek-forward-on-rebalance, monotonic control-offset merge, the transactional-id format, and
  best-effort stopCoordinator.

[kumarpritam863]

@bryanck can you please take a look into this.

[kumarpritam863]

@laskoviymishka can you please check this one.

[emlynazuma]

Alternative to the put() → merge(..., Long::max) line, in case it's useful — handling the rewind
in a ConsumerRebalanceListener so the replay never happens:

@Override public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
  for (TopicPartition partition : partitions) {
    Long inMemoryOffset = controlTopicOffsets.get(partition.partition());
    if (inMemoryOffset != null && inMemoryOffset > consumer.position(partition)) {
      consumer.seek(partition, inMemoryOffset);   // + a warn
    }
  }
}

merge(Long::max) stops the duplicate files, but the replay still happens: the replayed
DataComplete double-count in readyBuffer, so the commit still fires early and validThroughTs()
takes min(timestamp) over only the partitions that reported — the snapshot can claim a
valid-through-ts later than the true watermark. Seeking forward avoids that, and is safe because
the skipped records were consumed by this same Channel and are already in commitBuffer.

Context on #16282. Branch on top of 1.11.0 with regression tests:
apache-iceberg-1.11.0...emlynazuma:iceberg:tongwai/ikc-1.11.0-rebalance-fixes

[kumarpritam863]

@emlynazuma I am adding a proper rebalance consumer listener but I do not think it will be needed if the zombie coordinator scenario is properly handled which after the deterministic changes that I have made will make it almost impossible to have two coordinators running. But to be on the safer side I will add a proper rebalance listener. Also I think the best place to clear thinks are in the revoked partitions. As revoked path is called first if there is any think to revoke so we can just clear the Data Written, Data Complete and control topic offsets stored against those partitions. But I was just analyzing for it to not become complicated. Let me know about your thoughts on these.

[kumarpritam863]

I agree what you have suggested is simpler and more clean. Incorporated that change in the PR. Thanks.

[emlynazuma]

Looks like this landed in onPartitionsAssigned via seekToTrackedOffsets rather than clearing on
revoke — matches what I raised above about revoke-time clearing risking dropping buffered records
that haven't committed yet. And it's still doing work independent of the fencing: open()/close()
starting and stopping the coordinator is itself a -coord membership change, so the rewind can still
happen on a clean single-coordinator handoff, not just during a split-brain window.

Will test this against our repro setup and report back.

[kumarpritam863]

Yes @emlynazuma what you suggested looked clean and sufficient to handle seeking part. Also preventing zombie coordinator in the open and close is much more important than this as that is the reason which is causing this. Also I was thinking that rather than doing all this mumbo-zumbo can't we just do this:

protected void consumeAvailable(Duration pollDuration) {
  ConsumerRecords<String, byte[]> records = consumer.poll(pollDuration);
  while (!records.isEmpty()) {
    for (ConsumerRecord<String, byte[]> record : records) {
      // the consumer stores the offset of the next record to consume,
      // so increment the record offset by one
      if (record.offset() < controlTopicOffsets.getOrDefault(record.partition(), 0L)) {
        LOG.warn("Channel {} processed an already processed offset {} for partition {}", taskId, record.offset(), record.partition());
        continue;
      }
      controlTopicOffsets.put(record.partition(), record.offset() + 1);
      Event event = AvroUtil.decode(record.value());
      if (event.groupId().equals(connectGroupId)) {
        LOG.debug("Received event of type: {}", event.type().name());
        if (receive(new Envelope(event, record.partition(), record.offset()))) {
          LOG.info("Handled event of type: {}", event.type().name());
        }
      }
    }
    records = consumer.poll(pollDuration);
  }
}

This along with the open and close fix and ensuring closed coordinator in the committer should be sufficient I guess. What do you think?

[kumarpritam863]

I will make this change and test if this is suffcient as this will simplyfy and minimize the changes.