fishfishfishfishaa commented on PR #8671:
URL: https://github.com/apache/paimon/pull/8671#issuecomment-5102791720
## Normal scenarios
#### Writers reach EndInput at different checkpoints
An early EndInput writer keeps its `Long.MAX_VALUE` entry. It is treated as
already aligned for
later ordinary checkpoints, so the remaining writers can continue committing
ordinary data. The
final entry is committed only after all writers reach EndInput.
UT:
-
`CommittingWriteOperatorCoordinatorTest.testSubtasksEndInputAcrossDifferentCheckpoints`
#### Repeated EndInput before the final commit
EndInput may be replayed after recovery. The latest event is treated as the
authoritative state for
that writer and replaces the previous `Long.MAX_VALUE` entry. It does not
create a duplicate
ordinary checkpoint error.
UT:
-
`CommittingWriteOperatorCoordinatorTest.testRepeatedEndInputEventIsIdempotent`
- `WriterCommittablesTest.testRepeatedEndInputReplacesAuthoritativeEntry`
#### Repeated EndInput after the final commit
When `endInputCommitted=true`, the same coordinator ignores later EndInput
events and completed
checkpoint notifications.
UT:
-
`CommittingWriteOperatorCoordinatorTest.testRepeatedEndInputAfterFinalCommitIsIgnored`
## Failure scenarios
### WriteTask Failover
#### Writer task fails before EndInput
The coordinator remains in `RUNNING`. `subtaskReset` clears the failed
writer's cached
committables. The recovered writer replays its pending state and then
continues sending ordinary
checkpoint events.
UT:
-
`CommittingWriteOperatorCoordinatorTest.testPartialFailoverWithoutRestoring`
- `CommittingWriteOperatorCoordinatorTest.testPartialFailoverWithRestoring`
#### A writer task fails after EndInput
`subtaskReset` clears the failed writer's cached EndInput entry. When the
writer recovers, its
`RestoredCommittableEvent` may contain ordinary entries and a
`Long.MAX_VALUE` entry.
While the coordinator is still `RUNNING`:
- already completed ordinary entries are ignored;
- the restored EndInput entry is saved again;
- streaming waits for the other writers and a completed checkpoint; and
- checkpoint-disabled execution waits for the last writer and then commits
immediately.
UT:
-
`CommittingWriteOperatorCoordinatorTest.testPartialFailoverReplaysEndInputWhileCoordinatorIsRunning`
#### All writers EndInput but streaming is waiting for checkpoint completion
The EndInput entries are stored in writer operator state. After writer
recovery, the restore events
rebuild `allSubtasksEndInput()`. The coordinator still waits for a completed
checkpoint before the
final commit.
UT:
-
`CoordinatorCommittingRowDataStoreWriteOperatorTest.testPartialFailoverWhileAllSubtasksEndInputWaitForCheckpointComplete`
### JM Failover
#### JM failover with partial EndInput
The recreated coordinator enters `RESTORING` and waits for one restore event
from every writer.
All writers restore from the same completed checkpoint.
When only some writers contain EndInput:
1. ordinary entries are aligned at the restored checkpoint;
2. already committed ordinary entries are filtered or cleared;
3. the `Long.MAX_VALUE` entries remain buffered;
4. the coordinator changes to `RUNNING`; and
5. it waits for the remaining writers to reach EndInput.
UT:
-
`CommittingWriteOperatorCoordinatorTest.testGlobalRestoreRetainsPartialEndInput`
#### JM failover with all writers at EndInput
When all restore events contain EndInput, `allSubtasksEndInput()` becomes
true during recovery. The
recovery target changes to `Long.MAX_VALUE`, and the coordinator calls
`filterAndCommit`.
If recovery creates a missing snapshot, the configured recovery behavior
triggers another job
restart so that all writers restart from the new snapshot. If the snapshot
already exists,
`filterAndCommit` returns no new commit.
UT:
-
`CommittingWriteOperatorCoordinatorTest.testGlobalRestoreCommitsWhenAllSubtasksReachedEndInput`
#### Batch Mode JM failover without a completed checkpoint
Without a completed checkpoint, there is no writer pending state to restore.
Batch scheduling
recomputes the writer output. After all restarted writers send EndInput
again, the coordinator
calls `filterAndCommit(Long.MAX_VALUE)`.
UT&IT:
-
`CommittingWriteOperatorCoordinatorTest.testCheckpointDisabledCommitsWhenAllSubtasksEndInput`
- `CoordinatorEndInputCommitITCase.testCoordinatorCommitEndInputInBatchMode`
#### Writer task and JM failover happen together or in sequence
JM failover subsumes a writer region failover that is still running or has
just completed. In
Flink, each TaskExecutor observes the old JobMaster losing leadership,
disconnects that
JobManager connection, and calls `failExternally` for every task of the job.
The new JobMaster then
creates a new execution graph and globally restores every writer and the
coordinator from the same
completed checkpoint. This does not fail the TaskManager process itself; it
fails that job's tasks
running on the TaskManager. For each writer, restoring operator state is the
same mechanism as an
ordinary task failover, but the JM failover scope is global and all writers
use the same recovery
checkpoint. Therefore, a writer's local recovery attempt is not an
additional recovery authority:
- The old coordinator's in-memory cache, local `subtaskReset`, and replay
progress are discarded.
- If every checkpointed writer state contains EndInput, the restored
coordinator aligns all writer
events and uses `Long.MAX_VALUE` as the final recovery target.
- If only some checkpointed writer states contain EndInput, the restored
coordinator recovers the
ordinary checkpoint target and retains those `Long.MAX_VALUE` entries.
They are committed only
after the remaining writers reach EndInput and a later streaming
checkpoint completes.
UT:
-
`CommittingWriteOperatorCoordinatorTest.testWriterFailoverThenGlobalRestoreUsesCheckpointedEndInput`
-
`CommittingWriteOperatorCoordinatorTest.testWriterFailoverThenGlobalRestoreRetainsPartialCheckpointedEndInput`
### CK Abort
#### Checkpoint abort after partial EndInput
An aborted ordinary checkpoint does not remove pending committables. A later
completed checkpoint
commits accumulated ordinary entries. The early `Long.MAX_VALUE` entry
remains buffered until all
writers reach EndInput.
UT:
-
`CommittingWriteOperatorCoordinatorTest.testCheckpointAbortPreservesEarlyEndInput`
--
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]
