[BP-1.13][FLINK-22001] Fix forwarding of JobMaster exceptions to user

flink-clients/src/test/java/org/apache/flink/client/program/PerJobMiniClusterFactoryTest.java

@@ -113,23 +113,6 @@ public void testJobClientSavepoint() throws Exception {
                 () -> jobClient.stopWithSavepoint(true, null).get());
     }
 
-    @Test
-    public void testSubmissionError() throws Exception {
-        PerJobMiniClusterFactory perJobMiniClusterFactory = initializeMiniCluster();
-
-        // JobGraph is not a valid job
-
-        JobGraph jobGraph = JobGraphTestUtils.emptyJobGraph();
-
-        assertThrows(
-                "Could not instantiate JobManager",
-                ExecutionException.class,
-                () ->
-                        perJobMiniClusterFactory
-                                .submitJob(jobGraph, ClassLoader.getSystemClassLoader())
-                                .get());
-    }
-
     @Test
     public void testMultipleExecutions() throws Exception {
         PerJobMiniClusterFactory perJobMiniClusterFactory = initializeMiniCluster();

flink-core/src/main/java/org/apache/flink/api/common/JobStatus.java

@@ -24,8 +24,8 @@
 @PublicEvolving
 public enum JobStatus {
     /**
-     * The job has been received by the Dispatcher, and is waiting for the job manager to be
-     * created.
+     * The job has been received by the Dispatcher, and is waiting for the job manager to receive
+     * leadership and to be created.
      */
     INITIALIZING(TerminalState.NON_TERMINAL),
 

flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/JobDispatcherFactory.java

@@ -54,7 +54,7 @@ public MiniDispatcher createDispatcher(
                 fencingToken,
                 DispatcherServices.from(
                         partialDispatcherServicesWithJobGraphStore,
-                        DefaultJobManagerRunnerFactory.INSTANCE),
+                        JobMasterServiceLeadershipRunnerFactory.INSTANCE),
                 jobGraph,
                 dispatcherBootstrapFactory,
                 executionMode);