beryllw opened a new issue, #3543: URL: https://github.com/apache/fluss/issues/3543
### Search before asking - [x] I searched in the [issues](https://github.com/apache/fluss/issues) and found nothing similar. ### Fluss version 0.9.0 (latest release) ### Please describe the bug 🐞 For a running **partitioned Fluss source** with: - `scan.startup.mode = 'latest'` - `scan.partition.discovery.interval > 0` rows can be silently skipped if they are written to a **new partition before that partition is discovered** by the source enumerator. #### Expected behavior `scan.startup.mode=latest` should skip data that already existed **when the job started**, but it should not skip rows written to a partition that is created **after** the job is already running. #### Actual behavior For a newly created partition: 1. the partition is created and starts receiving rows 2. the running source does not see it until the next discovery cycle 3. when the partition is discovered, it is initialized from `latest` 4. rows already present in that partition are skipped forever #### Reproduction We could reproduced this with a new ITCase: - start a partitioned source with `scan.startup.mode='latest'` - wait for the initial discovery cycle - add a new partition - immediately write an early batch into that partition - wait for the next discovery cycle - write a late batch into the same partition Observed result: - only the late batch is consumed - the early batch is skipped #### Code path The issue comes from reusing the same offsets initializer for both: - partitions known at job startup - partitions discovered later Relevant code: - `fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkTableSource.java` - `fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumerator.java` `scan.startup.mode='latest'` is mapped to `OffsetsInitializer.latest()`, and newly discovered partitions are initialized through the same `startingOffsetsInitializer`, so they also start from the partition's latest offset at discovery time. #### Impact This can cause silent data loss for partitioned Flink jobs reading Fluss tables with `scan.startup.mode=latest`, especially when partitions are created dynamically and the first rows arrive before the next discovery tick. ### Solution Introduce a separate startup policy for **newly discovered partitions** instead of always reusing `scan.startup.mode`, or default `scan.new-partition.startup.mode=earliest` for **newly discovered partitions**. ### Are you willing to submit a PR? - [x] I'm willing to submit a PR! -- 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]
