jorgebo10 opened a new issue, #17637:
URL: https://github.com/apache/iceberg/issues/17637
### 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-<connector>-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 <connector>-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<TopicPartition>
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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
