@Override
    public void restartTasks(Set<ExecutionVertexID> verticesToRestart) {
        scheduledVertices.removeAll(verticesToRestart);
        maybeScheduleVertices(verticesToRestart);
    }
@Test
    void testRestartTasks() {
        final VertexwiseSchedulingStrategy schedulingStrategy =
                createSchedulingStrategy(testingSchedulingTopology);

        inputConsumableDecider.addSourceVertices(new HashSet<>(source));

        final Set<ExecutionVertexID> verticesToRestart =
                Stream.of(source, map, sink)
                        .flatMap(List::stream)
                        .map(TestingSchedulingExecutionVertex::getId)
                        .collect(Collectors.toSet());

        schedulingStrategy.restartTasks(verticesToRestart);

        final List<List<TestingSchedulingExecutionVertex>> expectedScheduledVertices =
                new ArrayList<>();
        expectedScheduledVertices.add(Collections.singletonList(source.get(0)));
        expectedScheduledVertices.add(Collections.singletonList(source.get(1)));
        assertLatestScheduledVerticesAreEqualTo(
                expectedScheduledVertices, testingSchedulerOperation);
    }