Revert "Capture initial JobStatus"

This reverts commit a63cd554e25a14f69adee3501d249a4129fa619f.

Author: liuml07

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

@@ -1648,15 +1648,6 @@ 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

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