[FLINK-22494][runtime] Refactors CheckpointsCleaner to handle also discardOnFailedStoring

Author: XComp

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

@@ -1209,20 +1209,7 @@ private void completePendingCheckpoint(PendingCheckpoint pendingCheckpoint)
                         completedCheckpoint, checkpointsCleaner, this::scheduleTriggerRequest);
             } catch (Exception exception) {
                 // we failed to store the completed checkpoint. Let's clean up
-                executor.execute(
-                        new Runnable() {
-                            @Override
-                            public void run() {
-                                try {
-                                    completedCheckpoint.discardOnFailedStoring();
-                                } catch (Throwable t) {
-                                    LOG.warn(
-                                            "Could not properly discard completed checkpoint {}.",
-                                            completedCheckpoint.getCheckpointID(),
-                                            t);
-                                }
-                            }
-                        });
+                checkpointsCleaner.cleanCheckpointOnFailedStoring(completedCheckpoint, executor);
 
                 sendAbortedMessages(
                         pendingCheckpoint.getCheckpointPlan().getTasksToCommitTo(),

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

@@ -18,6 +18,8 @@
 
 package org.apache.flink.runtime.checkpoint;
 
+import org.apache.flink.util.function.RunnableWithException;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -34,6 +36,7 @@
 @ThreadSafe
 public class CheckpointsCleaner implements Serializable {
     private static final Logger LOG = LoggerFactory.getLogger(CheckpointsCleaner.class);
+    private static final long serialVersionUID = 2545865801947537790L;
 
     private final AtomicInteger numberOfCheckpointsToClean;
 
@@ -50,23 +53,44 @@ public void cleanCheckpoint(
             boolean shouldDiscard,
             Runnable postCleanAction,
             Executor executor) {
+        cleanup(
+                checkpoint,
+                () -> {
+                    if (shouldDiscard) {
+                        checkpoint.discard();
+                    }
+                },
+                postCleanAction,
+                executor);
+    }
+
+    public void cleanCheckpointOnFailedStoring(
+            CompletedCheckpoint completedCheckpoint, Executor executor) {
+        cleanup(
+                completedCheckpoint,
+                completedCheckpoint::discardOnFailedStoring,
+                () -> {},
+                executor);
+    }
+
+    private void cleanup(
+            Checkpoint checkpoint,
+            RunnableWithException cleanupAction,
+            Runnable postCleanupAction,
+            Executor executor) {
         numberOfCheckpointsToClean.incrementAndGet();
         executor.execute(
                 () -> {
                     try {
-                        if (shouldDiscard) {
-                            try {
-                                checkpoint.discard();
-                            } catch (Exception e) {
-                                LOG.warn(
-                                        "Could not discard completed checkpoint {}.",
-                                        checkpoint.getCheckpointID(),
-                                        e);
-                            }
-                        }
+                        cleanupAction.run();
+                    } catch (Exception e) {
+                        LOG.warn(
+                                "Could not properly discard completed checkpoint {}.",
+                                checkpoint.getCheckpointID(),
+                                e);
                     } finally {
                         numberOfCheckpointsToClean.decrementAndGet();
-                        postCleanAction.run();
+                        postCleanupAction.run();
                     }
                 });
     }