Kafka Connect: duplicate coordinator after rebalance because close() re-derives leadership from an unstable consumer group

### Apache Iceberg version

1.11.0 (latest release)

### Query engine

Kafka Connect

### Please describe the bug 🐞

A single rebalance can leave two `CoordinatorThread`s alive for one connector. Both keep their own commit timer and both broadcast `StartCommit` on the control topic, which workers cannot distinguish from each other.

#### Symptom

Observed in production on a sink with 10 tasks: the connector's coordinator consumer group `connect--coord` had **two** members instead of the expected one, and stayed in `COMPLETING_REBALANCING` across repeated samples 60s apart. All 10 worker control groups (`cg-control-*`) were alive and stable at the same time.

When leadership cannot be resolved, the closing task logs:

```
WARN o.a.i.connect.channel.CommitterImpl - Committer -0 found no partitions assigned
across all members, cannot determine leader
```

#### Root cause

`CommitterImpl.close()` decides whether to stop its coordinator by re-deriving leadership from a live `Admin` describe of the connect consumer group (line numbers on current `main`):

```java
// CommitterImpl.java:179
if (hasLeaderPartition(closedPartitions)) {
LOG.info("Committer {} lost leader partition. Stopping coordinator.", taskId);
stopCoordinator();
}
```

```java
// CommitterImpl.java:78
boolean hasLeaderPartition(Collection currentAssignedPartitions) {
ConsumerGroupDescription groupDesc;
try (Admin admin = clientFactory.createAdmin()) {
groupDesc = KafkaUtils.consumerGroupDescription(config.connectGroupId(), admin);
}
...
}
```

But `close()` runs *during* the rebalance that is revoking the partitions. At that point the group is not stable, and `describeConsumerGroups` can return members that carry no assignments at all. `findFirstTopicPartition()` then returns `null`, `containsFirstPartition()` logs the warning above and returns `false`, and `stopCoordinator()` is never called.

The next `open()` on another task sees a settled group, finds it owns the lowest partition, and starts a coordinator of its own — `startCoordinator()`'s `coordinatorThread == null` guard is per-instance and cannot see the other task's thread. The previous coordinator is now orphaned: nothing holds a reference that would ever stop it.

The decision is also redundant. A task that owns a coordinator already knows it locally — `coordinatorThread != null` means "I started one and never stopped it". `close()` overrides that fact with an inference drawn from data that is specifically unreliable at that instant, and the inference's failure mode ("I cannot tell") resolves to "do nothing".

#### Reproduction

Deterministic, in-process, no broker required — simulate one rebalance across two committers and count live `iceberg-coord` threads:

1. Group is stable; task A owns the lowest partition, `open()` elects it and starts one coordinator.
2. Group description switches to members with empty assignments (the mid-rebalance state).
3. `taskA.close([lowest partition])` — returns without stopping the coordinator.
4. Group description settles again; `taskB.open([lowest partition])` elects B and starts a second coordinator.

```
java.lang.AssertionError: after one rebalance there must still be exactly one live iceberg-coord thread
Expected size: 1 but was: 2 in:
[Thread[#36,iceberg-coord,5,main], Thread[#35,iceberg-coord,5,main]]
```

#### Suggested fix

Stop asking the broker on the revocation path. Record the partition leadership was won with at `open()` time, and have `close()` compare it against the revoked partitions locally:

```java
private TopicPartition leaderPartition; // set in hasLeaderPartition() when the check passes

// close()
if (coordinatorThread != null
&& (leaderPartition == null || closedPartitions.contains(leaderPartition))) {
stopCoordinator();
}
```

Checking the *specific* elected partition rather than "did I lose anything" keeps this correct under cooperative rebalancing, where a task can be closed for a subset of its partitions and legitimately keep leadership. Falling back to stopping when the elected partition is unknown is the safe direction, since an orphaned coordinator corrupts data while a missing one is re-elected on the next `open()`.

I have a branch with the fix and the test above, and will open a PR referencing this issue.

#### Related but distinct

- #16016 — same symptom class (a coordinator thread that outlives the task) but a different trigger and mechanism: a write failure kills the task, and `CoordinatorThread.terminate()` does not join the thread, so the coordinator keeps running against a catalog that `IcebergSinkTask.close()` has already closed. Not addressed by this fix.
- #17340 — duplicate data files, but attributed there to `Channel.consumeAvailable` regressing the control-topic watermark on re-read, which is independent of coordinator count.

### Willingness to contribute

- [x] I can contribute a fix for this bug independently
- [ ] I would be willing to contribute a fix for this bug with guidance from the Iceberg community
- [ ] I cannot contribute a fix for this bug at this time

Start in CommitterImpl.java with open(), hasLeaderPartition(), and close(), then follow the coordinator lifecycle through the described rebalance reproduction. Verify the change by simulating empty assignments during close and confirming that after the next open there is exactly one live iceberg-coord thread; the issue author’s branch and planned PR indicate work is already underway.

Written by the indexing model from the issue text.