zhuzhurk commented on a change in pull request #9920: [FLINK-14389][runtime] Restore task state before restarting tasks in DefaultScheduler URL: https://github.com/apache/flink/pull/9920#discussion_r339682365
########## File path: flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/DefaultSchedulerTest.java ########## @@ -388,6 +394,127 @@ public void vertexIsNotAffectedByOutdatedDeployment() { assertThat(sv1.getState(), is(equalTo(ExecutionState.SCHEDULED))); } + @Test + public void abortPendingCheckpointsWhenRestartingTasks() { + final JobGraph jobGraph = singleNonParallelJobVertexJobGraph(); + enableCheckpointing(jobGraph); + + final DefaultScheduler scheduler = createSchedulerAndStartScheduling(jobGraph); + + final ArchivedExecutionVertex onlyExecutionVertex = Iterables.getOnlyElement(scheduler.requestJob().getAllExecutionVertices()); + final ExecutionAttemptID attemptId = onlyExecutionVertex.getCurrentExecutionAttempt().getAttemptId(); + scheduler.updateTaskExecutionState(new TaskExecutionState(jobGraph.getJobID(), attemptId, ExecutionState.RUNNING)); + + final CheckpointCoordinator checkpointCoordinator = getCheckpointCoordinator(scheduler); + + checkpointCoordinator.triggerCheckpoint(System.currentTimeMillis(), false); + assertThat(checkpointCoordinator.getNumberOfPendingCheckpoints(), is(equalTo(1))); + + scheduler.updateTaskExecutionState(new TaskExecutionState(jobGraph.getJobID(), attemptId, ExecutionState.FAILED)); + taskRestartExecutor.triggerScheduledTasks(); + assertThat(checkpointCoordinator.getNumberOfPendingCheckpoints(), is(equalTo(0))); + } + + @Test + public void restoreStateWhenRestartingTasks() throws Exception { + final JobGraph jobGraph = singleNonParallelJobVertexJobGraph(); + enableCheckpointing(jobGraph); + + final DefaultScheduler scheduler = createSchedulerAndStartScheduling(jobGraph); + + final ArchivedExecutionVertex onlyExecutionVertex = Iterables.getOnlyElement(scheduler.requestJob().getAllExecutionVertices()); + final ExecutionAttemptID attemptId = onlyExecutionVertex.getCurrentExecutionAttempt().getAttemptId(); + scheduler.updateTaskExecutionState(new TaskExecutionState(jobGraph.getJobID(), attemptId, ExecutionState.RUNNING)); + + final CheckpointCoordinator checkpointCoordinator = getCheckpointCoordinator(scheduler); + + // register a stateful master hook to help verify state restore + final TestMasterHook masterHook = new TestMasterHook("testHook"); + checkpointCoordinator.addMasterHook(masterHook); + + // complete one checkpoint for state restore + checkpointCoordinator.triggerCheckpoint(System.currentTimeMillis(), false); + acknowledgeCheckpoint(checkpointCoordinator, jobGraph.getJobID(), attemptId); + + scheduler.updateTaskExecutionState(new TaskExecutionState(jobGraph.getJobID(), attemptId, ExecutionState.FAILED)); + taskRestartExecutor.triggerScheduledTasks(); + assertThat(masterHook.getRestoreCount(), is(equalTo(1))); + } + + @Test + public void failGlobalWhenRestoringStateFails() throws Exception { + final JobGraph jobGraph = singleNonParallelJobVertexJobGraph(); + final JobVertex onlyJobVertex = getOnlyJobVertex(jobGraph); + enableCheckpointing(jobGraph); + + final DefaultScheduler scheduler = createSchedulerAndStartScheduling(jobGraph); + + final ArchivedExecutionVertex onlyExecutionVertex = Iterables.getOnlyElement(scheduler.requestJob().getAllExecutionVertices()); + final ExecutionAttemptID attemptId = onlyExecutionVertex.getCurrentExecutionAttempt().getAttemptId(); + scheduler.updateTaskExecutionState(new TaskExecutionState(jobGraph.getJobID(), attemptId, ExecutionState.RUNNING)); + + final CheckpointCoordinator checkpointCoordinator = getCheckpointCoordinator(scheduler); + + // register a master hook to fail state restore + final TestMasterHook masterHook = new TestMasterHook("testHook"); + masterHook.enableFailOnRestore(); + checkpointCoordinator.addMasterHook(masterHook); + + // complete one checkpoint for state restore + checkpointCoordinator.triggerCheckpoint(System.currentTimeMillis(), false); + acknowledgeCheckpoint(checkpointCoordinator, jobGraph.getJobID(), attemptId); + + scheduler.updateTaskExecutionState(new TaskExecutionState(jobGraph.getJobID(), attemptId, ExecutionState.FAILED)); + taskRestartExecutor.triggerScheduledTasks(); + final List<ExecutionVertexID> deployedExecutionVertices = testExecutionVertexOperations.getDeployedVertices(); + + // the first task failover should be skipped on state restore failure + final ExecutionVertexID executionVertexId = new ExecutionVertexID(onlyJobVertex.getID(), 0); + assertThat(deployedExecutionVertices, contains(executionVertexId)); + + // a global failure should be triggered on state restore failure + masterHook.disableFailOnRestore(); + taskRestartExecutor.triggerScheduledTasks(); + assertThat(deployedExecutionVertices, contains(executionVertexId, executionVertexId)); + } + + private void acknowledgeCheckpoint( + final CheckpointCoordinator checkpointCoordinator, + final JobID jobID, + final ExecutionAttemptID attemptId) throws Exception { Review comment: Ok. done by changing it to `acknowledgePendingCheckpoint(scheduler, checkpointId)` to acknowledge all the executions for the given checkpoint. ---------------------------------------------------------------- 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services