spoorthibasu commented on PR #4360:
URL: https://github.com/apache/flink-cdc/pull/4360#issuecomment-4190249987
Thanks for the detailed review, @fcfangcc, this was a very helpful catch.
You’re right there were two issues here:
1. Original bug:
Schema changes could split writes within a checkpoint, but all batches were
committed into a single Iceberg snapshot with the same sequence number, so
equality-deletes could not suppress earlier data.
Fixed by flushing per-table on schema change and committing each batch as a
separate snapshot.
2. Parallelism:
• Writer: flushTableWriter could return early and cause
batchIndex drift across subtasks. Fixed by advancing the counter before the
early return.
• Committer: Wrappers were committed individually, so same
batchIndex data from different subtasks could produce multiple snapshots. Fixed
by grouping by batchIndex and merging into a single snapshot.
Retry safety:
Only the final non-empty batch writes flink-max-committed-checkpoint-id.
Intermediate batches carry flink.batch-index and flink.checkpoint-id, so
retries resume from the next uncommitted batch instead of skipping.
The changes are in this commit for reference:
https://github.com/apache/flink-cdc/pull/4360/changes/d42c29021a4bc8f2d20e140b73373aa0472eb08a
I also looked at your implementation. Using explicit per-snapshot markers to
track batch boundaries makes sense. One concern is retry behavior: if the
committer crashes after batch 0 but before batch 1, the retry would see the
checkpoint id property on the batch 0 snapshot and skip the entire checkpoint,
leaving batch 1’s data uncommitted.
In this approach, only the final non-empty batch sets that property, so
retries resume from the next uncommitted batch instead of skipping the
checkpoint entirely.
Happy to discuss further, and I would appreciate your feedback on this
approach.
--
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]
