@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 testLargeMultiplier() throws Exception {
        final int parallelism = 3;
        final JobVertex jobVertex = createNoOpVertex(parallelism);
        final ExecutionGraph executionGraph = createExecutionGraph(jobVertex);

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

        Thread.sleep(10);

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

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

        // no task can exceed the large baseline
        assertThat(slowTasks).isEmpty();
    }