morningman opened a new issue, #716: URL: https://github.com/apache/doris-flink-connector/issues/716
### Search before asking - [x] I had searched in the [issues](https://github.com/apache/doris-flink-connector/issues?q=is%3Aissue) and found no similar issues. ### Version master @ 9550f9a0ebb16d95964ec65340f877b120f15403 (2026-11-05). ### What's Wrong? With `source.changelog.offset-table` and `source.changelog.consumer-id`, the source can publish an offset before the sink has committed the data that precedes it: - At `snapshotState(checkpointId)` the enumerator records the last window that all readers finished, and in `notifyCheckpointComplete(checkpointId)` it publishes that window to the offset table ([DorisSourceEnumerator.java#L178-L215](https://github.com/apache/doris-flink-connector/blob/9550f9a0ebb16d95964ec65340f877b120f15403/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/source/enumerator/DorisSourceEnumerator.java#L178-L215)). - A Doris sink with 2PC pre-commits at the same checkpoint and commits from its committer after the checkpoint completes ([DorisCommitter.java#L94](https://github.com/apache/doris-flink-connector/blob/9550f9a0ebb16d95964ec65340f877b120f15403/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/sink/committer/DorisCommitter.java#L94)). - The two completion notifications are independent, and nothing orders the offset publication after the sink commit. If the job stops after the offset is written but before the sink transaction is committed, and the job is then restarted without Flink state (the case the offset table exists for), two things happen: - the new writer aborts the lingering pre-committed transactions for its label prefix ([DorisWriter.java#L120-L167](https://github.com/apache/doris-flink-connector/blob/9550f9a0ebb16d95964ec65340f877b120f15403/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/sink/writer/DorisWriter.java#L120-L167)); - the source resumes from the published offset. The rows of that last checkpoint are then neither committed nor read again. The same applies to other sinks that commit after checkpoint completion, such as a transactional Kafka sink. ### What You Expected? A stateless restart from the offset table never skips data that was not committed downstream. Possible approaches: - publish the offset of checkpoint N only after a later checkpoint completes (one checkpoint of lag); - document that the offset table is safe only with sinks that commit at the barrier; - let the sink write the offset in the same transaction as the data. ### How to Reproduce? 1. Run Doris → Flink → Doris with a Duplicate Key target, `sink.enable-2pc = true`, and an offset table configured on the source. 2. Kill the TaskManager right after a checkpoint completes: after the enumerator has written the offset, before the committer commits the transaction. 3. Restart the job without a savepoint or checkpoint. 4. Compare the source and target row counts for the affected window. Suspected: the rows of the last checkpoint are missing. ### Anything Else? Found by code reading while documenting the incremental source; not yet reproduced. A Doris sink without 2PC, which is the default for Unique Key targets, commits each checkpoint's stream load at the barrier, so the window is much narrower there. ### Are you willing to submit PR? - [ ] Yes I am willing to submit a PR! ### Code of Conduct - [x] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct) -- 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]
