tongwai-wong-appier commented on issue #16282:
URL: https://github.com/apache/iceberg/issues/16282#issuecomment-5082896086
Here's the timeline you asked for. No `seek()` is involved — all it takes is
a member entering or
leaving `<group>-coord`, and a zombie coordinator is exactly such a member.
Two rebalances to keep apart:
- **source side** — the sink tasks' consumer group over the source topic
partitions
- **coord side** — the coordinator's own consumer group `<group>-coord` over
the control topic
They're coupled: the coordinator runs on whichever task owns the lowest
source partition
(`containsFirstPartition` → `findFirstTopicPartition`), so every source-side
rebalance that moves
that partition starts or stops a coordinator — i.e. changes coord-side
membership. And source-side
rebalances flap.
Take d1 → d2 → d1, a single flap of the lowest source partition:
- **d1 loses it.** `close(closedPartitions)` →
`hasLeaderPartition(closedPartitions)` describes the
group to find the first partition; during a rebalance that snapshot can
come back empty or
partial, so it returns false and `stopCoordinator()` is skipped.
**Coordinator C keeps running.**
- **d2 gains it.** `open()` → `startCoordinator()` → **C' is constructed** →
its first `poll()` sends
JoinGroup → coord-side rebalance.
- **d1 gains it back.** `open()` → `startCoordinator()`, but `if (null ==
this.coordinatorThread)`
is false, so it's a **no-op**. C carries straight on, with its
`commitBuffer`, `currentCommitId`
and `controlTopicOffsets` intact.
- **d2 loses it.** `close()` → `stopCoordinator()` → `consumer.close()` →
LeaveGroup → coord-side
rebalance again.
Across the whole flap C is never reconstructed — the no-op guard keeps it
alive — while C' joins and
leaves around it. IKC never sets `partition.assignment.strategy`, so the
client default
`[RangeAssignor, CooperativeStickyAssignor]` takes its protocol from the
first entry: **eager**.
Eager revokes every partition and discards its fetch position, even for a
member that gets the same
partition straight back; `updateFetchPositions()` then refetches the
committed offset. That's the
rewind, and it happens inside the consumer — which is why there's no
`.seek()` to find.
Nothing on that path touches C's in-memory state, and no
`ConsumerRebalanceListener` is registered.
That's t2: `commitBuffer` 0–7, `controlTopicOffsets` `{0:8}`, position back
at 0. The replayed
records carry the current `commitId` (the group offset only advances on a
successful `doCommit()`),
so `isCommitReady()` fires mid-drain at 8+2 = 10 and the commit runs while
the map is still `{0:2}`.
Your co-reset invariant holds for C' — the Coordinator that was constructed.
C is the one that
breaks, and it was never constructed.
--
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]
