tongwai-wong-appier commented on issue #16282:
URL: https://github.com/apache/iceberg/issues/16282#issuecomment-5087157214
We took a different fix on 1.11.0. Two parts.
**Rewind** — a `ConsumerRebalanceListener` that seeks forward instead of
letting the replay happen:
```java
@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
}
}
}
```
This does strictly more than `merge(..., Long::max)`. That keeps the floor
from regressing, but the
replay still happens — the replayed `DataComplete` still double-count in
`readyBuffer`, so the commit
still fires early. No duplicate files, but `validThroughTs()` then takes
`min(timestamp)` over only
the partitions that did report, and the ones that didn't aren't in the list
to make
`allMatch(timestamp != null)` false, so the snapshot claims a
`valid-through-ts` later than the true
watermark. Seeking forward means no replay at all, so buffer / offsets map /
ready count stay
consistent. Skipping past the group's committed offset is safe because those
records were consumed
by this same `Channel` — their envelopes are in `commitBuffer` and get
committed from there.
**Zombie** — we kept the existing election but made *de-election*
deterministic: record the leader
partition at election time and have `close()` check
`closedPartitions.contains(leaderPartition)`,
instead of re-running `hasLeaderPartition()`. That re-query is the thing
that returns an empty or
partial member snapshot mid-rebalance and leaves the coordinator running.
Much smaller than your
rework and it removes the same root cause; your `save()`-level reconcile is
the stronger version
since it also self-heals missed callbacks.
Two notes on the PR:
- `put()` → `merge(..., Long::max)` is an independent fix for #16282 — worth
saying so in the
description so it isn't dropped as unrelated to the election rework.
- `isCommitReady()` still compares `receivedPartitionCount`, which counts
duplicates. You fixed the
right-hand side; that's the left half of the same quorum.
`validThroughTs()` also doesn't filter
`readyBuffer` by `commitId` while `isCommitReady()` does.
Branch with both plus regression tests:
https://github.com/emlynazuma/iceberg/tree/tongwai/ikc-1.11.0-rebalance-fixes
Haven't run your PR against our repro setup yet — will report back when I do.
--
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]
