Repository: apache/flink
File: flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java (Java)

Code before:
                                                Events.JobStatusChangeEvent.name()
                                                        .equals(event.getName()))
                                .map(Event::getAttributes)
                                .map(x -> x.get("newJobStatus")))
                .containsExactly(
                        JobStatus.RUNNING.toString(),
                        JobStatus.FAILING.toString(),
                        JobStatus.FAILED.toString());

        assertThat(
                        jobEvents.stream()
                                .filter(
                                        event ->
                                                Events.AllSubtasksStatusChangeEvent.name()
                                                        .equals(event.getName()))
                                .map(Event::getAttributes)
                                .map(
                                        x ->
                                                x.get(
                                                        AllSubTasksRunningOrFinishedStateTimeMetrics
                                                                .STATUS_ATTRIBUTE)))
                .containsExactly(
                        ALL_RUNNING_OR_FINISHED.toString(), NOT_ALL_RUNNING_OR_FINISHED.toString());
    }

    @Test
    void testJobFailureWhenTaskExecutorHeartbeatTimeout() throws Exception {
        final List<Event> jobEvents = new ArrayList<>();
        final TestingHeartbeatServices testingHeartbeatService =
                new TestingHeartbeatServices(heartbeatInterval, heartbeatTimeout);

        runJobFailureWhenTaskExecutorTerminatesTest(
                testingHeartbeatService,
                (localTaskManagerLocation, jobMasterGateway) ->
                        testingHeartbeatService.triggerHeartbeatTimeout(
                                jmResourceId, localTaskManagerLocation.getResourceID()),
                jobEvents);
        assertThat(
                        jobEvents.stream()
                                .filter(
                                        event ->
                                                Events.JobStatusChangeEvent.name()
                                                        .equals(event.getName()))
                                .map(Event::getAttributes)
                                .map(x -> x.get("newJobStatus")))
                .containsExactly(
                        JobStatus.RUNNING.toString(),
                        JobStatus.FAILING.toString(),
                        JobStatus.FAILED.toString());

        assertThat(

Review comment (suggestion):
Sorry for the delay in responding. It looks like we have got to a the root cause of this, in that we have 2 different types of scheduler that contain different job status's.

I see that this method  checks for the schedulerType and performs different asserts. I cannot see from the test that this method is driven for each scheduler type. If the 2 callers of this method cover the 2 scheduler types - it would be good if this was more explicit in the code.



Code after:
        assertJobStatusTransitions(schedulerType, jobEvents);
    }

    @Test
    void testJobFailureWhenTaskExecutorHeartbeatTimeout() throws Exception {
        final List<Event> jobEvents = new ArrayList<>();
        final TestingHeartbeatServices testingHeartbeatService =
                new TestingHeartbeatServices(heartbeatInterval, heartbeatTimeout);

        final SchedulerType schedulerType =
                runJobFailureWhenTaskExecutorTerminatesTest(
                        testingHeartbeatService,
                        (localTaskManagerLocation, jobMasterGateway) ->
                                testingHeartbeatService.triggerHeartbeatTimeout(
                                        jmResourceId, localTaskManagerLocation.getResourceID()),
                        jobEvents);

        assertJobStatusTransitions(schedulerType, jobEvents);
    }

    /**
     * Asserts that job status transitions are as expected based on the scheduler type.
     * DefaultScheduler does not emit CREATED state, while AdaptiveScheduler and
     * AdaptiveBatchScheduler do.
     */
    private static void assertJobStatusTransitions(
            SchedulerType schedulerType, List<Event> jobEvents) {
        final List<String> jobStatusTransitions =
                jobEvents.stream()
                        .filter(event -> Events.JobStatusChangeEvent.name().equals(event.getName()))
                        .map(Event::getAttributes)
                        .map(x -> (String) x.get("newJobStatus"))
                        .collect(Collectors.toList());

        if (schedulerType == SchedulerType.Adaptive) {
            // Adaptive schedulers emit CREATED: CREATED → RUNNING → FAILING → FAILED
            assertThat(jobStatusTransitions)
                    .containsExactly(
                            JobStatus.CREATED.toString(),
                            JobStatus.RUNNING.toString(),
                            JobStatus.FAILING.toString(),
                            JobStatus.FAILED.toString());
        } else {
            // Default scheduler does not emit CREATED: RUNNING → FAILING → FAILED
            assertThat(jobStatusTransitions)
                    .containsExactly(
                            JobStatus.RUNNING.toString(),
                            JobStatus.FAILING.toString(),
                            JobStatus.FAILED.toString());
        }