zhongyujiang opened a new pull request, #16950: URL: https://github.com/apache/iceberg/pull/16950
## Summary Port of #16929 (Flink 2.1) to Flink 1.20 and Flink 2.0. `DataIterator.seek()` overwrites `fileOffset` after `updateCurrentIterator()` has correctly advanced it past files skipped by residual filters. This causes checkpoint to record a wrong file offset, leading to duplicate consumption or `IllegalStateException` on recovery. **The fix:** 1. Increment `fileOffset` in the skip-files loop so it tracks position as files are skipped 2. Remove the incorrect `fileOffset`/`recordOffset` assignments at the end of `seek()` The bug and fix are identical to #16929 — this PR simply applies the same change to the remaining Flink versions that share the same `DataIterator` code. ## Tests - `testDataIteratorWithResidualFilter`: validates that `fileOffset` matches the actual file being read when a residual filter is applied - `testInitializationWithHeadFilesSkipped`: directly asserts that `fileOffset = 1` after `seek(0, 0)` when file 0 is skipped by residual filter - `testSeekResumesCorrectlyAfterHeadFilesSkipped`: checkpoint/restore regression test — verifies that after skipping head files, checkpoint position is correct and restore reads the right data without duplicates cc @pvary Could you please review? This is the same fix from #16929 ported to Flink 1.20 and 2.0. -- 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]
