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

[FLINK-XXXX] Draft separation of leader election and creation of JobMasterService

[FLINK-XXXX] Continued work on JobMasterServiceLeadershipRunner

[FLINK-XXXX] Integrate RunningJobsRegistry, Cancelling state and termination future watching

[FLINK-XXXX] Delete old JobManagerRunnerImpl classes

[FLINK-22001] Add tests for DefaultJobMasterServiceProcess

[FLINK-22001][hotfix] Clean up ITCase a bit

[FLINK-22001] Add missing check for empty job graph

[FLINK-22001] Rename JobMasterServiceFactoryNg to JobMasterServiceFactory

This closes #15715.

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);