Capture initial JobStatus

Author: liuml07

flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraph.java

@@ -1648,6 +1648,15 @@ public void updateAccumulators(AccumulatorSnapshot accumulatorSnapshot) {
     public void registerJobStatusListener(JobStatusListener listener) {
         if (listener != null) {
             jobStatusListeners.add(listener);
+            // Emit current state to the newly registered listener
+            // This ensures listeners don't miss the initial state
+            try {
+                listener.jobStatusChanges(getJobID(), state, stateTimestamps[state.ordinal()]);
+            } catch (Throwable t) {
+                LOG.warn(
+                        "Error while notifying newly registered JobStatusListener of current state",
+                        t);
+            }
         }
     }
 

flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java

@@ -1714,7 +1714,8 @@ void testJobFailureWhenGracefulTaskExecutorTermination() throws Exception {
                                                         .equals(event.getName()))
                                 .map(Event::getAttributes)
                                 .map(x -> x.get("newJobStatus")))
-                .containsSubsequence(
+                .containsExactly(
+                        JobStatus.CREATED.toString(),
                         JobStatus.RUNNING.toString(),
                         JobStatus.FAILING.toString(),
                         JobStatus.FAILED.toString());
@@ -1755,7 +1756,8 @@ void testJobFailureWhenTaskExecutorHeartbeatTimeout() throws Exception {
                                                         .equals(event.getName()))
                                 .map(Event::getAttributes)
                                 .map(x -> x.get("newJobStatus")))
-                .containsSubsequence(
+                .containsExactly(
+                        JobStatus.CREATED.toString(),
                         JobStatus.RUNNING.toString(),
                         JobStatus.FAILING.toString(),
                         JobStatus.FAILED.toString());