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

Code before:
        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.Default) {
            // Default scheduler does not emit CREATED: RUNNING → FAILING → FAILED
            assertThat(jobStatusTransitions)
                    .containsExactly(
                            JobStatus.RUNNING.toString(),
                            JobStatus.FAILING.toString(),
                            JobStatus.FAILED.toString());
        } else {
            // Adaptive schedulers emit CREATED: CREATED → RUNNING → FAILING → FAILED
            assertThat(jobStatusTransitions)
                    .containsExactly(
                            JobStatus.CREATED.toString(),
                            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 ->

Review comment (suggestion):
I prefer using `if (schedulerType == SchedulerType.Adaptive)` as `Default` is a more general scheduler.

Code after:
        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());
        }

        assertThat(
                        jobEvents.stream()
                                .filter(
                                        event ->
                                                Events.AllSubtasksStatusChangeEvent.name()
                                                        .equals(event.getName()))
                                .map(Event::getAttributes)
                                .map(
                                        x ->