ifndef-SleePy commented on PR #8671: URL: https://github.com/apache/paimon/pull/8671#issuecomment-5193374211
Before diving into implementation details, I'd like to discuss the design from a high-level view first. #### Race condition issue: The original `CommitterOperator` design: - For batch jobs, it commits directly in `endInput`. - For streaming with checkpointing enabled, it commits in `notifyCheckpointComplete` of the final checkpoint. This PR instead always triggers the commit directly on end input. For the streaming + checkpointing-enabled case, this differs from `CommitterOperator`, and I think the race condition the reviewer mentioned is not the low-level RPC ordering issue — it's that there are now two flows that can trigger a commit: end input and checkpoint completion. Although we handle them on a single thread and use the checkpoint id to distinguish them, and Paimon provides `filterAndCommit` to tolerate the overlap, this still feels weird: end input may commit the previous checkpoint's data ahead of time, before `notifyCheckpointComplete` is triggered, and that checkpoint is not even guaranteed to succeed. I'm not sure whether this has other side effects; it's a very subtle situation, and it gets more complex once we consider non-idempotent operators. There is also the global failover case: the in-memory `endInputCommitted` variable is gone, so the behavior differs again — the coordinator will re-attempt both the checkpoint-driven and the end-input-driven commits. This should be filtered out by `filterAndCommit` and not actually re-commit, but it is again a new, different path. Another case: if a writer calls `endInput` first and a normal checkpoint is triggered afterwards, the `Long.MAX_VALUE` end-input committable gets persisted into that writer's operator state. If a global failover then happens, on restore the writer reports a `RestoredCommittableEvent` with a normal (restore) checkpoint id, but its payload still carries the `Long.MAX_VALUE` committable. This feels odd, because the original design seems to expect that a restore only carries committables whose checkpoint id is less than or equal to the restore checkpoint id. I haven't verified whether this actually leads to an incorrect commit (the coordinator's `headMap`-based selection may exclude `Long.MAX_VALUE` anyway), but carrying it under a normal restore checkpoint id breaks the contract the code otherwise assumes. #### Failover on finished task: For a writer task that has already sent its end input, is there any scenario where it would not be brought up again to re-send `RestoredCommittableEvent` (e.g. deployed as `FINISHED_ON_RESTORE`)? My current understanding is that we should never reach `FINISHED_ON_RESTORE` in this scenario, so `RestoredCommittableEvent` is always re-sent, but I haven't verified this with a test. I'm not sure whether you've considered and tested this scenario. -- 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]
