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_r339682440
##########
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 {
+
+ final long checkpointId =
checkpointCoordinator.getPendingCheckpoints().keySet().iterator().next();
+ final AcknowledgeCheckpoint acknowledgeCheckpoint = new
AcknowledgeCheckpoint(
+ jobID,
+ attemptId,
+ checkpointId);
+
checkpointCoordinator.receiveAcknowledgeMessage(acknowledgeCheckpoint, "Unknown
location");
+ }
+
+ private void enableCheckpointing(final JobGraph jobGraph) {
+ final JobVertex onlyJobVertex = getOnlyJobVertex(jobGraph);
Review comment:
Agreed. done.
----------------------------------------------------------------
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:
[email protected]
With regards,
Apache Git Services
