@VisibleForTesting
    Map<ExecutionVertexID, Collection<ExecutionAttemptID>> findSlowTasks(
            final ExecutionGraph executionGraph) {
        final long currentTimeMillis = System.currentTimeMillis();

        final Map<ExecutionVertexID, Collection<ExecutionAttemptID>> slowTasks = new HashMap<>();

        final List<ExecutionJobVertex> jobVerticesToCheck = getJobVerticesToCheck(executionGraph);

        for (ExecutionJobVertex ejv : jobVerticesToCheck) {
            final ExecutionTimeWithInputBytes baseline = getBaseline(ejv, currentTimeMillis);

            for (ExecutionVertex ev : ejv.getTaskVertices()) {
                if (ev.getExecutionState().isTerminal()) {
                    continue;
                }

                final List<ExecutionAttemptID> slowExecutions =
                        findExecutionsExceedingBaseline(
                                ev.getCurrentExecutions(), baseline, currentTimeMillis);

                if (!slowExecutions.isEmpty()) {
                    slowTasks.put(ev.getID(), slowExecutions);
                }
            }
        }

        return slowTasks;
    }
@Test
    void testFinishedTaskExceedRatioInDynamicGraph() throws Exception {
        final int parallelism = 3;
        final JobVertex jobVertex1 = createNoOpVertex(parallelism);
        // create jobVertex2 and leave its parallelism unset
        final JobVertex jobVertex2 = new JobVertex("vertex2");
        jobVertex2.setInvokableClass(NoOpInvokable.class);
        jobVertex2.connectNewDataSetAsInput(
                jobVertex1, DistributionPattern.ALL_TO_ALL, ResultPartitionType.BLOCKING);
        final ExecutionGraph executionGraph = createDynamicExecutionGraph(jobVertex1, jobVertex2);

        final ExecutionTimeBasedSlowTaskDetector slowTaskDetector =
                createSlowTaskDetector(0.3, 1, 0);

        final ExecutionVertex ev13 =
                executionGraph.getJobVertex(jobVertex1.getID()).getTaskVertices()[2];
        ev13.getCurrentExecutionAttempt().markFinished();

        final Map<ExecutionVertexID, Collection<ExecutionAttemptID>> slowTasks =
                slowTaskDetector.findSlowTasks(executionGraph);

        assertThat(slowTasks).hasSize(2);
    }