[FLINK-11813] Introduces cleanup of dirty jobs in Dispatcher

flink-clients/src/main/java/org/apache/flink/client/deployment/application/ApplicationDispatcherGatewayServiceFactory.java

@@ -26,11 +26,13 @@
 import org.apache.flink.runtime.dispatcher.DispatcherFactory;
 import org.apache.flink.runtime.dispatcher.DispatcherId;
 import org.apache.flink.runtime.dispatcher.PartialDispatcherServices;
-import org.apache.flink.runtime.dispatcher.PartialDispatcherServicesWithJobGraphStore;
+import org.apache.flink.runtime.dispatcher.PartialDispatcherServicesWithJobPersistenceComponents;
 import org.apache.flink.runtime.dispatcher.runner.AbstractDispatcherLeaderProcess;
 import org.apache.flink.runtime.dispatcher.runner.DefaultDispatcherGatewayService;
+import org.apache.flink.runtime.highavailability.JobResultStore;
 import org.apache.flink.runtime.jobgraph.JobGraph;
 import org.apache.flink.runtime.jobmanager.JobGraphWriter;
+import org.apache.flink.runtime.jobmaster.JobResult;
 import org.apache.flink.runtime.rpc.RpcService;
 import org.apache.flink.util.FlinkRuntimeException;
 
@@ -83,7 +85,9 @@ public ApplicationDispatcherGatewayServiceFactory(
     public AbstractDispatcherLeaderProcess.DispatcherGatewayService create(
             DispatcherId fencingToken,
             Collection<JobGraph> recoveredJobs,
-            JobGraphWriter jobGraphWriter) {
+            Collection<JobResult> globallyTerminatedJobs,
+            JobGraphWriter jobGraphWriter,
+            JobResultStore jobResultStore) {
 
         final List<JobID> recoveredJobIds = getRecoveredJobIds(recoveredJobs);
 
@@ -94,6 +98,7 @@ public AbstractDispatcherLeaderProcess.DispatcherGatewayService create(
                             rpcService,
                             fencingToken,
                             recoveredJobs,
+                            globallyTerminatedJobs,
                             (dispatcherGateway, scheduledExecutor, errorHandler) ->
                                     new ApplicationDispatcherBootstrap(
                                             application,
@@ -102,8 +107,8 @@ public AbstractDispatcherLeaderProcess.DispatcherGatewayService create(
                                             dispatcherGateway,
                                             scheduledExecutor,
                                             errorHandler),
-                            PartialDispatcherServicesWithJobGraphStore.from(
-                                    partialDispatcherServices, jobGraphWriter));
+                            PartialDispatcherServicesWithJobPersistenceComponents.from(
+                                    partialDispatcherServices, jobGraphWriter, jobResultStore));
         } catch (Exception e) {
             throw new FlinkRuntimeException("Could not create the Dispatcher rpc endpoint.", e);
         }

flink-clients/src/main/java/org/apache/flink/client/deployment/application/ApplicationDispatcherLeaderProcessFactoryFactory.java

@@ -26,7 +26,7 @@
 import org.apache.flink.runtime.dispatcher.runner.DispatcherLeaderProcessFactory;
 import org.apache.flink.runtime.dispatcher.runner.DispatcherLeaderProcessFactoryFactory;
 import org.apache.flink.runtime.dispatcher.runner.SessionDispatcherLeaderProcessFactory;
-import org.apache.flink.runtime.jobmanager.JobGraphStoreFactory;
+import org.apache.flink.runtime.jobmanager.JobPersistenceComponentFactory;
 import org.apache.flink.runtime.rpc.FatalErrorHandler;
 import org.apache.flink.runtime.rpc.RpcService;
 
@@ -59,7 +59,7 @@ private ApplicationDispatcherLeaderProcessFactoryFactory(
 
     @Override
     public DispatcherLeaderProcessFactory createFactory(
-            JobGraphStoreFactory jobGraphStoreFactory,
+            JobPersistenceComponentFactory jobPersistenceComponentFactory,
             Executor ioExecutor,
             RpcService rpcService,
             PartialDispatcherServices partialDispatcherServices,
@@ -74,7 +74,10 @@ public DispatcherLeaderProcessFactory createFactory(
                         partialDispatcherServices);
 
         return new SessionDispatcherLeaderProcessFactory(
-                dispatcherServiceFactory, jobGraphStoreFactory, ioExecutor, fatalErrorHandler);
+                dispatcherServiceFactory,
+                jobPersistenceComponentFactory,
+                ioExecutor,
+                fatalErrorHandler);
     }
 
     public static ApplicationDispatcherLeaderProcessFactoryFactory create(

flink-core/src/main/java/org/apache/flink/util/function/QuintFunction.java

@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.util.function;
+
+import org.apache.flink.annotation.PublicEvolving;
+
+/**
+ * Function which takes three arguments.
+ *
+ * @param <S> type of the first argument
+ * @param <T> type of the second argument
+ * @param <U> type of the third argument
+ * @param <V> type of the fourth argument
+ * @param <W> type of the fifth argument
+ * @param <R> type of the return value
+ */
+@PublicEvolving
+@FunctionalInterface
+public interface QuintFunction<S, T, U, V, W, R> {
+
+    /**
+     * Applies this function to the given arguments.
+     *
+     * @param s the first function argument
+     * @param t the second function argument
+     * @param u the third function argument
+     * @param v the fourth function argument
+     * @param w the fifth function argument
+     * @return the function result
+     */
+    R apply(S s, T t, U u, V v, W w);
+}

flink-core/src/main/java/org/apache/flink/util/function/QuintFunctionWithException.java

@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.util.function;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.util.ExceptionUtils;
+
+/**
+ * Function which takes five arguments.
+ *
+ * @param <S> type of the first argument
+ * @param <T> type of the second argument
+ * @param <U> type of the third argument
+ * @param <V> type of the four argument
+ * @param <W> type of the five argument
+ * @param <R> type of the return value
+ * @param <E> type of the thrown exception
+ */
+@PublicEvolving
+@FunctionalInterface
+public interface QuintFunctionWithException<S, T, U, V, W, R, E extends Throwable> {
+
+    /**
+     * Applies this function to the given arguments.
+     *
+     * @param s the first function argument
+     * @param t the second function argument
+     * @param u the third function argument
+     * @param v the four function argument
+     * @param w the five function argument
+     * @return the function result
+     * @throws E if it fails
+     */
+    R apply(S s, T t, U u, V v, W w) throws E;
+
+    /**
+     * Convert at {@link QuintFunctionWithException} into a {@link QuintFunction}.
+     *
+     * @param quintFunctionWithException function with exception to convert into a function
+     * @param <A> first input type
+     * @param <B> second input type
+     * @param <C> third input type
+     * @param <D> fourth input type
+     * @param <E> fifth input type
+     * @param <R> output type
+     * @return {@link QuintFunction} which throws all checked exception as an unchecked exception.
+     */
+    static <A, B, C, D, E, R> QuintFunction<A, B, C, D, E, R> unchecked(
+            QuintFunctionWithException<A, B, C, D, E, R, ?> quintFunctionWithException) {
+        return (A a, B b, C c, D d, E e) -> {
+            try {
+                return quintFunctionWithException.apply(a, b, c, d, e);
+            } catch (Throwable t) {
+                ExceptionUtils.rethrow(t);
+                // we need this to appease the compiler :-(
+                return null;
+            }
+        };
+    }
+}

flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesCheckpointRecoveryFactory.java

@@ -71,8 +71,7 @@ public KubernetesCheckpointRecoveryFactory(
 
     @Override
     public CompletedCheckpointStore createRecoveredCompletedCheckpointStore(
-            JobID jobID, int maxNumberOfCheckpointsToRetain, ClassLoader userClassLoader)
-            throws Exception {
+            JobID jobID, int maxNumberOfCheckpointsToRetain) throws Exception {
 
         final String configMapName = getConfigMapNameFunction.apply(jobID);
         return KubernetesUtils.createCompletedCheckpointStore(

flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobServer.java

@@ -846,7 +846,8 @@ public boolean cleanupJob(JobID jobId, boolean cleanupBlobStoreFiles) {
             try {
                 FileUtils.deleteDirectory(jobDir);
 
-                // NOTE: Instead of going through blobExpiryTimes, keep lingering entries - they
+                // NOTE on why blobExpiryTimes are not cleaned up:
+                //       Instead of going through blobExpiryTimes, keep lingering entries - they
                 //       will be cleaned up by the timer task which tolerates non-existing files
                 //       If inserted again with the same IDs (via put()), the TTL will be updated
                 //       again.

flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointRecoveryFactory.java

@@ -19,6 +19,10 @@
 package org.apache.flink.runtime.checkpoint;
 
 import org.apache.flink.api.common.JobID;
+import org.apache.flink.configuration.CheckpointingOptions;
+import org.apache.flink.configuration.Configuration;
+
+import org.slf4j.Logger;
 
 /** A factory for per Job checkpoint recovery components. */
 public interface CheckpointRecoveryFactory {
@@ -30,12 +34,43 @@ public interface CheckpointRecoveryFactory {
      *
      * @param jobId Job ID to recover checkpoints for
      * @param maxNumberOfCheckpointsToRetain Maximum number of checkpoints to retain
-     * @param userClassLoader User code class loader of the job
      * @return {@link CompletedCheckpointStore} instance for the job
      */
     CompletedCheckpointStore createRecoveredCompletedCheckpointStore(
-            JobID jobId, int maxNumberOfCheckpointsToRetain, ClassLoader userClassLoader)
-            throws Exception;
+            JobID jobId, int maxNumberOfCheckpointsToRetain) throws Exception;
+
+    /**
+     * Instantiates the {@link CompletedCheckpointStore} based on the passed {@code Configuration}.
+     *
+     * @param jobId The {@code JobID} for which the {@code CompletedCheckpointStore} shall be
+     *     created.
+     * @param config The {@code Configuration} that shall be used (see {@link
+     *     CheckpointingOptions#MAX_RETAINED_CHECKPOINTS}.
+     * @param logger The logger that shall be used internally.
+     * @return The {@code CompletedCheckpointStore} instance for the given {@code Job}.
+     * @throws Exception if an error occurs while instantiating the {@code
+     *     CompletedCheckpointStore}.
+     */
+    default CompletedCheckpointStore createRecoveredCompletedCheckpointStore(
+            JobID jobId, Configuration config, Logger logger) throws Exception {
+        int maxNumberOfCheckpointsToRetain =
+                config.getInteger(CheckpointingOptions.MAX_RETAINED_CHECKPOINTS);
+
+        if (maxNumberOfCheckpointsToRetain <= 0) {
+            // warning and use 1 as the default value if the setting in
+            // state.checkpoints.max-retained-checkpoints is not greater than 0.
+            logger.warn(
+                    "The setting for '{} : {}' is invalid. Using default value of {}",
+                    CheckpointingOptions.MAX_RETAINED_CHECKPOINTS.key(),
+                    maxNumberOfCheckpointsToRetain,
+                    CheckpointingOptions.MAX_RETAINED_CHECKPOINTS.defaultValue());
+
+            maxNumberOfCheckpointsToRetain =
+                    CheckpointingOptions.MAX_RETAINED_CHECKPOINTS.defaultValue();
+        }
+
+        return this.createRecoveredCompletedCheckpointStore(jobId, maxNumberOfCheckpointsToRetain);
+    }
 
     /**
      * Creates a {@link CheckpointIDCounter} instance for a job.

flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PerJobCheckpointRecoveryFactory.java

@@ -68,7 +68,7 @@ public PerJobCheckpointRecoveryFactory(
 
     @Override
     public CompletedCheckpointStore createRecoveredCompletedCheckpointStore(
-            JobID jobId, int maxNumberOfCheckpointsToRetain, ClassLoader userClassLoader) {
+            JobID jobId, int maxNumberOfCheckpointsToRetain) {
         return store.compute(
                 jobId,
                 (key, previous) ->

flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/StandaloneCheckpointRecoveryFactory.java

@@ -26,8 +26,7 @@ public class StandaloneCheckpointRecoveryFactory implements CheckpointRecoveryFa
 
     @Override
     public CompletedCheckpointStore createRecoveredCompletedCheckpointStore(
-            JobID jobId, int maxNumberOfCheckpointsToRetain, ClassLoader userClassLoader)
-            throws Exception {
+            JobID jobId, int maxNumberOfCheckpointsToRetain) throws Exception {
 
         return new StandaloneCompletedCheckpointStore(maxNumberOfCheckpointsToRetain);
     }

flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/ZooKeeperCheckpointRecoveryFactory.java

@@ -47,8 +47,7 @@ public ZooKeeperCheckpointRecoveryFactory(
 
     @Override
     public CompletedCheckpointStore createRecoveredCompletedCheckpointStore(
-            JobID jobId, int maxNumberOfCheckpointsToRetain, ClassLoader userClassLoader)
-            throws Exception {
+            JobID jobId, int maxNumberOfCheckpointsToRetain) throws Exception {
 
         return ZooKeeperUtils.createCompletedCheckpoints(
                 ZooKeeperUtils.useNamespaceAndEnsurePath(

flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/ApplicationStatus.java

@@ -19,6 +19,10 @@
 package org.apache.flink.runtime.clusterframework;
 
 import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.flink.shaded.guava30.com.google.common.collect.BiMap;
+import org.apache.flink.shaded.guava30.com.google.common.collect.HashBiMap;
 
 /** The status of an application. */
 public enum ApplicationStatus {
@@ -37,6 +41,16 @@ public enum ApplicationStatus {
 
     // ------------------------------------------------------------------------
 
+    private static final BiMap<JobStatus, ApplicationStatus> JOB_STATUS_APPLICATION_STATUS_BI_MAP =
+            HashBiMap.create();
+
+    static {
+        // only globally-terminated JobStatus have a corresponding ApplicationStatus
+        JOB_STATUS_APPLICATION_STATUS_BI_MAP.put(JobStatus.FAILED, ApplicationStatus.FAILED);
+        JOB_STATUS_APPLICATION_STATUS_BI_MAP.put(JobStatus.CANCELED, ApplicationStatus.CANCELED);
+        JOB_STATUS_APPLICATION_STATUS_BI_MAP.put(JobStatus.FINISHED, ApplicationStatus.SUCCEEDED);
+    }
+
     /** The associated process exit code */
     private final int processExitCode;
 
@@ -59,20 +73,20 @@ public int processExitCode() {
      * #UNKNOWN}.
      */
     public static ApplicationStatus fromJobStatus(JobStatus jobStatus) {
-        if (jobStatus == null) {
+        if (jobStatus == null || !JOB_STATUS_APPLICATION_STATUS_BI_MAP.containsKey(jobStatus)) {
             return UNKNOWN;
-        } else {
-            switch (jobStatus) {
-                case FAILED:
-                    return FAILED;
-                case CANCELED:
-                    return CANCELED;
-                case FINISHED:
-                    return SUCCEEDED;
-
-                default:
-                    return UNKNOWN;
-            }
         }
+
+        return JOB_STATUS_APPLICATION_STATUS_BI_MAP.get(jobStatus);
+    }
+
+    public static JobStatus fromApplicationStatus(ApplicationStatus applicationStatus) {
+        Preconditions.checkNotNull(applicationStatus, "ApplicationStatus must not be null");
+        if (JOB_STATUS_APPLICATION_STATUS_BI_MAP.inverse().containsKey(applicationStatus)) {
+            throw new UnsupportedOperationException(
+                    applicationStatus.name() + " cannot be mapped to a JobStatus.");
+        }
+
+        return JOB_STATUS_APPLICATION_STATUS_BI_MAP.inverse().get(applicationStatus);
     }
 }