@Override
    public Set<ExecutionVertexID> getTasksNeedingRestart(
            ExecutionVertexID executionVertexId, Throwable cause) {
        return IterableUtils.toStream(topology.getVertices())
                .map(SchedulingExecutionVertex::getId)
                .collect(Collectors.toSet());
    }
@Test
    void testGetTasksNeedingRestart() {
        final TestingSchedulingTopology topology = new TestingSchedulingTopology();

        final TestingSchedulingExecutionVertex v1 = topology.newExecutionVertex();
        final TestingSchedulingExecutionVertex v2 = topology.newExecutionVertex();
        final TestingSchedulingExecutionVertex v3 = topology.newExecutionVertex();

        topology.connect(v1, v2, ResultPartitionType.PIPELINED);
        topology.connect(v2, v3, ResultPartitionType.BLOCKING);

        final RestartAllFailoverStrategy strategy = new RestartAllFailoverStrategy(topology);

        assertThat(new HashSet<>(Arrays.asList(v1.getId(), v2.getId(), v3.getId())))
                .isEqualTo(
                        strategy.getTasksNeedingRestart(v1.getId(), new Exception("Test failure")));
    }