fishfishfishfishaa commented on PR #8671: URL: https://github.com/apache/paimon/pull/8671#issuecomment-5151776127
> There is a race condition between `END_INPUT` and checkpoint completion. If the checkpoint completes first and `END_INPUT` is not received until later, the final event will not record and reuse this completed checkpoint; furthermore, a bounded job may stop generating checkpoints, resulting in the final batch of data never being committed. The current tests rely on manually triggering another checkpoint and do not cover the actual race condition. Good catch. `END_INPUT` is sent through `sendEventToCoordinator`, while checkpoint completion is delivered through `notifyCheckpointComplete`. Although both ultimately enter the same JobMaster `RpcEndpoint`, the former uses an ask-style RPC and the latter a tell-style RPC. They therefore have different actor senders and do not satisfy Pekko’s same-sender FIFO condition, so their delivery order is not guaranteed. Even with direct final commit after all `END_INPUT` events are collected, `alignCommittables` may still fail and trigger failover if checkpoint completion is processed before the `END_INPUT` event reaches the coordinator. The current implementation addresses this in two ways: 1. In streaming mode, the coordinator performs an idempotent final commit immediately after collecting all `END_INPUT` events, without waiting for another checkpoint. 2. The writer persists pending END_INPUT committables in operator state. If the race triggers failover, the new coordinator restores and commits them through `RestoredCommittableEvent`, preventing data loss. I added two ITs: - `testFinalCheckpointCompletionBeforeEndInputEventCommitsAfterFailover` deterministically reproduces the strict race by controlling checkpoint timing and delaying the `END_INPUT` event. It verifies that a non-empty END_INPUT committable is restored and committed after failover. Manual checkpoints are used only to control ordering, not as a commit dependency. - `testLateEndInputAfterCheckpointCompletionIsCommitted` uses periodic checkpoints and disables `execution.checkpointing.checkpoints-after-tasks-finish`. It verifies the final result and proves that the final batch is committed even when no later checkpoint is generated. The current implementation ensures that the final commit is not lost. The race may trigger failover, but it does not affect consistency. -- 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]
