vbhanuchander-lang commented on issue #17340: URL: https://github.com/apache/iceberg/issues/17340#issuecomment-5639629082
@akomisarek good question — that filter is narrower than it looks, in three ways. **It is keyed on what *that table's* snapshot already recorded.** `committedOffsets` comes from `lastCommittedOffsetsForTable(table, branch)`, i.e. the `kafka.connect.offsets.<topic>.<group>` property on that table's current snapshot. So the guard only exists for a table+partition pair that has already committed at least once. The `minOffset == null` branch is a pass, not a skip: for a partition with no recorded entry — a newly added table, a table that has never seen envelopes from that control-topic partition, or the first commits after the connector is created — every envelope goes through, replays included. **It can only be as far forward as the value that was stored.** The value written back is `max(committedOffsets, controlTopicOffsets)` per partition, so for a partition that already has an entry it cannot regress. But where there is no prior entry, whatever `controlTopicOffsets` holds at that moment is stamped in directly. That is the part #17933 was about: `consumeAvailable` recorded the consumed position with an unconditional `put(partition, offset + 1)`, so a re-read of the control topic moved the tracked position **backwards**. A regressed value then both (a) becomes the group offset a restarted channel resumes from, which is what produces the re-read in the first place, and (b) can be the value stamped onto a snapshot for a partition with no prior entry. Afterwards, replayed envelopes are compared against a watermark that is behind them and pass. **Nothing downstream catches what slips past.** `distinctByKey(ContentFile::location)` dedupes only *within* a single commit's payload list, and `appendFiles` does no path-level comparison against what the table already references. So one envelope through the filter is one duplicate data file in the table — which is why the report shows double-referenced files rather than an error. Worth adding that the filter also cannot see a second coordinator. If two `CoordinatorThread`s are alive for one connector (#17637), each reads the table's committed offsets at its own moment and both can pass the same envelopes. That is a separate cause with the same symptom, and the offset arithmetic here does not help with it. So: the filter is a real guard, but it is a per-table low-water mark that only protects a partition it already knows about, and it was being fed a watermark that could move backwards. Whether your specific duplicates came through the `null` branch or a regressed watermark would show in the `kafka.connect.offsets.*` property on the snapshots around the incident — if the recorded position for the affected partition ever decreases between consecutive snapshots, that is the second case. -- 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]
