tongwai-wong-appier commented on issue #16282:
URL: https://github.com/apache/iceberg/issues/16282#issuecomment-5077459104
Thanks @kumarpritam863 — you're right, and the mechanism I wrote in the
issue body is wrong. The
floor check has been there since #10351, so it's in 1.10.1 too, and it does
drop the stale
`DATA_WRITTEN` in the case I described.
But could the recorded offset itself go stale? `controlTopicOffsets` is
overwritten rather than
max-merged, so it moves *backwards* whenever the coordinator's control-topic
consumer rewinds:
```java
// Channel.java
controlTopicOffsets.put(record.partition(), record.offset() + 1); // plain
put()
```
and that map — not the envelopes being committed — is what the snapshot's
floor is derived from:
```java
// Coordinator.java
Map<Integer, Long> mergedOffsets =
Stream.of(committedOffsets, controlTopicOffsets)
.flatMap(map -> map.entrySet().stream())
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue,
Long::max));
```
Single control-topic partition, commit expects 10 partitions, table floor
starts at 1. Offsets are
as stored by the code (`record.offset() + 1`):
| | position | `controlTopicOffsets` | in buffer | ready count | result |
|---|---|---|---|---|---|
| t1 | consumed 1–8 | `{0:9}` | 1–8 | 8 | not ready |
| t2 | **rewinds to 1** | `{0:9}` | 1–8 | 8 | — |
| t3 | re-reads 1, 2 | **`{0:3}`** | 1–8 | 8+2 = **10** → commit | floor is
1, so envelopes 1–8 are appended, but the summary records `max(1, 3) = 3` |
| t4 | reads 3–10 | `{0:11}` | 3–10 | — | — |
| t5 | next commit | | 3–20 | | floor is 3, so 3–20 all pass → files from
3–8 appended again |
The snapshot at t3 holds files up to control offset 8 but advertises a floor
of 3. Once the
recorded floor lags the snapshot's own contents, the filter can't drop those
events on the next
cycle — and `commitConsumerOffsets()` reads the same map, so the group
offset is pushed backwards
too, keeping the window open.
`validateWith(offsetValidator(...))` from #14510 doesn't seem to close this:
at t5 both
`expectedOffsets` and `lastCommittedOffsets(baseSnapshots)` are `3`, so
validation passes. It
guards against a concurrent writer during retry, not against a floor that
was already written
behind its own snapshot.
On how the rewind gets triggered:
- It needs neither two coordinators nor the early commit at t3 —
`isCommitTimedOut() ->
commit(true)` in the same window has the same effect.
- The `-coord` consumer runs with `enable.auto.commit=false`, so any rejoin
resets its position to
the last `commitConsumerOffsets()`.
Does this hold, or am I missing a guard?
--
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]
