@Test public void testRestartTasks() { JobVertexID jobVertexID = new JobVertexID(); testingSchedulingTopology.addSchedulingExecutionVertex(new TestingSchedulingExecutionVertex(jobVertexID, 0)); testingSchedulingTopology.addSchedulingExecutionVertex(new TestingSchedulingExecutionVertex(jobVertexID, 1)); testingSchedulingTopology.addSchedulingExecutionVertex(new TestingSchedulingExecutionVertex(jobVertexID, 2)); testingSchedulingTopology.addSchedulingExecutionVertex(new TestingSchedulingExecutionVertex(jobVertexID, 3)); testingSchedulingTopology.addSchedulingExecutionVertex(new TestingSchedulingExecutionVertex(jobVertexID, 4)); Set<ExecutionVertexID> verticesToRestart1 = new HashSet<>(Arrays.asList( new ExecutionVertexID(jobVertexID, 0), new ExecutionVertexID(jobVertexID, 4))); schedulingStrategy.restartTasks(verticesToRestart1); Set<ExecutionVertexID> verticesToRestart2 = new HashSet<>(Arrays.asList( new ExecutionVertexID(jobVertexID, 1), new ExecutionVertexID(jobVertexID, 2), new ExecutionVertexID(jobVertexID, 3))); schedulingStrategy.restartTasks(verticesToRestart2); assertThat(testingSchedulerOperations.getScheduledVertices(), hasSize(2)); Collection<ExecutionVertexDeploymentOption> scheduledVertices1 = testingSchedulerOperations.getScheduledVertices().get(0); assertThat(getExecutionVertexIdsFromDeployOptions(scheduledVertices1), containsInAnyOrder(verticesToRestart1.toArray())); Collection<ExecutionVertexDeploymentOption> scheduledVertices2 = testingSchedulerOperations.getScheduledVertices().get(1); assertThat(getExecutionVertexIdsFromDeployOptions(scheduledVertices2), containsInAnyOrder(verticesToRestart2.toArray())); }
@Override public void restartTasks(Set<ExecutionVertexID> verticesToRestart) { allocateSlotsAndDeploy(verticesToRestart); }
EagerSchedulingStrategy implements SchedulingStrategy { @Override public void restartTasks(Set<ExecutionVertexID> verticesToRestart) { allocateSlotsAndDeploy(verticesToRestart); } }
EagerSchedulingStrategy implements SchedulingStrategy { @Override public void restartTasks(Set<ExecutionVertexID> verticesToRestart) { allocateSlotsAndDeploy(verticesToRestart); } EagerSchedulingStrategy(
			SchedulerOperations schedulerOperations,
			SchedulingTopology schedulingTopology); }
EagerSchedulingStrategy implements SchedulingStrategy { @Override public void restartTasks(Set<ExecutionVertexID> verticesToRestart) { allocateSlotsAndDeploy(verticesToRestart); } EagerSchedulingStrategy(
			SchedulerOperations schedulerOperations,
			SchedulingTopology schedulingTopology); @Override void startScheduling(); @Override void restartTasks(Set<ExecutionVertexID> verticesToRestart); @Override void onExecutionStateChange(ExecutionVertexID executionVertexId, ExecutionState executionState); @Override void onPartitionConsumable(ExecutionVertexID executionVertexId, ResultPartitionID resultPartitionId); }
EagerSchedulingStrategy implements SchedulingStrategy { @Override public void restartTasks(Set<ExecutionVertexID> verticesToRestart) { allocateSlotsAndDeploy(verticesToRestart); } EagerSchedulingStrategy(
			SchedulerOperations schedulerOperations,
			SchedulingTopology schedulingTopology); @Override void startScheduling(); @Override void restartTasks(Set<ExecutionVertexID> verticesToRestart); @Override void onExecutionStateChange(ExecutionVertexID executionVertexId, ExecutionState executionState); @Override void onPartitionConsumable(ExecutionVertexID executionVertexId, ResultPartitionID resultPartitionId);  }