[FLINK-11813] Introduces basic JobManagerRunner implementation

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

@@ -25,9 +25,12 @@
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.JobManagerOptions;
 import org.apache.flink.configuration.SecurityOptions;
+import org.apache.flink.runtime.dispatcher.cleanup.GloballyCleanableResource;
+import org.apache.flink.runtime.dispatcher.cleanup.LocallyCleanableResource;
 import org.apache.flink.runtime.net.SSLUtils;
 import org.apache.flink.util.ExceptionUtils;
 import org.apache.flink.util.FileUtils;
+import org.apache.flink.util.FlinkException;
 import org.apache.flink.util.NetUtils;
 import org.apache.flink.util.ShutdownHookUtil;
 
@@ -71,7 +74,12 @@
  * the directory structure to store the BLOBs or temporarily cache them.
  */
 public class BlobServer extends Thread
-        implements BlobService, BlobWriter, PermanentBlobService, TransientBlobService {
+        implements BlobService,
+                BlobWriter,
+                PermanentBlobService,
+                TransientBlobService,
+                LocallyCleanableResource,
+                GloballyCleanableResource {
 
     /** The log object used for debugging. */
     private static final Logger LOG = LoggerFactory.getLogger(BlobServer.class);
@@ -824,46 +832,58 @@ public boolean deletePermanent(JobID jobId, PermanentBlobKey key) {
     }
 
     /**
-     * Removes all BLOBs from local and HA store belonging to the given job ID.
+     * Deletes locally stored artifacts for the job represented by the given {@link JobID}. This
+     * doesn't touch the job's entry in the {@link BlobStore} to enable recovering.
      *
-     * @param jobId ID of the job this blob belongs to
-     * @param cleanupBlobStoreFiles True if the corresponding blob store files shall be cleaned up
-     *     as well. Otherwise false.
-     * @return <tt>true</tt> if the job directory is successfully deleted or non-existing;
-     *     <tt>false</tt> otherwise
+     * @param jobId The {@code JobID} of the job that is subject to cleanup.
+     * @throws IOException if the cleanup failed.
      */
-    public boolean cleanupJob(JobID jobId, boolean cleanupBlobStoreFiles) {
+    @Override
+    public void localCleanup(JobID jobId) throws IOException {
         checkNotNull(jobId);
 
         final File jobDir =
                 new File(BlobUtils.getStorageLocationPath(storageDir.getAbsolutePath(), jobId));
 
+        FileUtils.deleteDirectory(jobDir);
+
+        // 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.
+    }
+
+    /**
+     * Removes all BLOBs from local and HA store belonging to the given {@link JobID}.
+     *
+     * @param jobId ID of the job this blob belongs to
+     * @throws Exception if the cleanup fails.
+     */
+    @Override
+    public void globalCleanup(JobID jobId) throws Exception {
+        checkNotNull(jobId);
+
         readWriteLock.writeLock().lock();
 
         try {
-            // delete locally
-            boolean deletedLocally = false;
-            try {
-                FileUtils.deleteDirectory(jobDir);
-
-                // 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.
+            Exception exception = null;
 
-                deletedLocally = true;
+            try {
+                localCleanup(jobId);
             } catch (IOException e) {
-                LOG.warn(
-                        "Failed to locally delete BLOB storage directory at "
-                                + jobDir.getAbsolutePath(),
-                        e);
+                exception = e;
             }
 
-            // delete in HA blob store files
-            final boolean deletedHA = !cleanupBlobStoreFiles || blobStore.deleteAll(jobId);
+            if (!blobStore.deleteAll(jobId)) {
+                exception =
+                        ExceptionUtils.firstOrSuppressed(
+                                new FlinkException(
+                                        "Error while cleaning up the BlobStore for job " + jobId),
+                                exception);
+            }
 
-            return deletedLocally && deletedHA;
+            ExceptionUtils.tryRethrowException(exception);
         } finally {
             readWriteLock.writeLock().unlock();
         }

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

@@ -156,7 +156,13 @@ private boolean delete(String blobPath) {
 
             Path path = new Path(blobPath);
 
-            boolean result = fileSystem.delete(path, true);
+            boolean result = true;
+            if (fileSystem.exists(path)) {
+                result = fileSystem.delete(path, true);
+            } else {
+                LOG.debug(
+                        "The given path {} is not present anymore. No deletion is required.", path);
+            }
 
             // send a call to delete the directory containing the file. This will
             // fail (and be ignored) when some files still exist.

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

@@ -26,16 +26,16 @@
 /** The status of an application. */
 public enum ApplicationStatus {
 
-    /** Application finished successfully */
+    /** Application finished successfully. */
     SUCCEEDED(0),
 
-    /** Application encountered an unrecoverable failure or error */
+    /** Application encountered an unrecoverable failure or error. */
     FAILED(1443),
 
-    /** Application was canceled or killed on request */
+    /** Application was canceled or killed on request. */
     CANCELED(0),
 
-    /** Application status is not known */
+    /** Application status is not known. */
     UNKNOWN(1445);
 
     // ------------------------------------------------------------------------
@@ -50,15 +50,15 @@ public enum ApplicationStatus {
         JOB_STATUS_APPLICATION_STATUS_BI_MAP.put(JobStatus.FINISHED, ApplicationStatus.SUCCEEDED);
     }
 
-    /** The associated process exit code */
+    /** The associated process exit code. */
     private final int processExitCode;
 
-    private ApplicationStatus(int exitCode) {
+    ApplicationStatus(int exitCode) {
         this.processExitCode = exitCode;
     }
 
     /**
-     * Gets the process exit code associated with this status
+     * Gets the process exit code associated with this status.
      *
      * @return The associated process exit code.
      */