vbhanuchander-lang opened a new pull request, #17933:
URL: https://github.com/apache/iceberg/pull/17933
Closes #17340.
Opening this at @ericyangliu's invitation on the issue — the diagnosis and
the production evidence
(149 double-referenced files, ~112k duplicated rows) are his.
## The bug
`Channel.consumeAvailable` recorded the consumed position with an
unconditional `put`:
```java
controlTopicOffsets.put(record.partition(), record.offset() + 1);
```
Nothing compares against the value already stored, so any re-read of a
control topic partition —
a rebalance resuming the consumer from the last committed group offsets, as
in the report — moves
the tracked position **backwards**.
That regression is durable rather than transient, because the map is not
just bookkeeping:
- `commitConsumerOffsets()` commits it for the consumer group, so the next
restart resumes from the
regressed offset and re-reads more.
- `Coordinator.commitToTable` merges it into `kafka.connect.offsets` on the
snapshot, which is the
watermark the min-offset filter uses on subsequent commits.
Once the watermark is behind, replayed `DataWritten` envelopes pass that
filter. `distinctByKey`
only dedupes within one commit and append does no path-level dedup, so the
same data files are
committed again and every scan reads them twice.
## The change
One line: keep the highest position seen for the partition.
```java
controlTopicOffsets.merge(record.partition(), record.offset() + 1,
Long::max);
```
This is what every reader of `controlTopicOffsets()` already assumes —
`commitToTable` even folds
it in with `Long::max` against the last committed offsets. Making the map
itself monotonic is
consistent with that, and it does not change the offsets recorded on the
forward path.
## Tests
`TestChannel` drives a `Channel` over a `MockConsumer`. Consuming offsets
0-4 reaches a watermark
of 5; a seek back to 1 then delivers a partial replay ending at offset 2,
which is the shape of the
re-read in the report.
- `controlTopicOffsetsTrackTheHighestPositionConsumed` — the map stays at 5.
With the fix reverted
it is 3.
- `committedControlTopicOffsetsDoNotRegressOnReplay` — asserts what the
channel actually commits to
Kafka, `OffsetAndMetadata{offset=5}`. With the fix reverted it commits 3,
which is the offset a
restarted channel would resume from.
- `controlTopicOffsetsAreTrackedPerPartition` — partitions stay independent.
Both regression tests fail on `main` without the change. Full module suite
passes (136 tests), as
do `spotlessCheck` and checkstyle.
## Not included
The issue also floats a bounded set of recently committed file locations in
the coordinator as a
content-level backstop for replays the offset arithmetic cannot see. That is
a design call for
maintainers and a larger change, so I have left it out rather than hold up
the correctness fix.
Happy to follow up if it is wanted.
--
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]
