[FLINK-22494][runtime] Adds PossibleInconsistentStateException handling to CheckpointCoordinator

Author: XComp

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

@@ -37,6 +37,7 @@
 import org.apache.flink.runtime.messages.checkpoint.DeclineCheckpoint;
 import org.apache.flink.runtime.operators.coordination.OperatorCoordinator;
 import org.apache.flink.runtime.operators.coordination.OperatorInfo;
+import org.apache.flink.runtime.persistence.PossibleInconsistentStateException;
 import org.apache.flink.runtime.state.CheckpointStorage;
 import org.apache.flink.runtime.state.CheckpointStorageCoordinatorView;
 import org.apache.flink.runtime.state.CheckpointStorageLocation;
@@ -1208,8 +1209,16 @@ private void completePendingCheckpoint(PendingCheckpoint pendingCheckpoint)
                 completedCheckpointStore.addCheckpoint(
                         completedCheckpoint, checkpointsCleaner, this::scheduleTriggerRequest);
             } catch (Exception exception) {
-                // we failed to store the completed checkpoint. Let's clean up
-                checkpointsCleaner.cleanCheckpointOnFailedStoring(completedCheckpoint, executor);
+                if (exception instanceof PossibleInconsistentStateException) {
+                    LOG.warn(
+                            "An error occurred while writing checkpoint {} to the underlying metadata store. Flink was not able to determine whether the metadata was successfully persisted. The corresponding state located at '{}' won't be discarded and needs to be cleaned up manually.",
+                            completedCheckpoint.getCheckpointID(),
+                            completedCheckpoint.getExternalPointer());
+                } else {
+                    // we failed to store the completed checkpoint. Let's clean up
+                    checkpointsCleaner.cleanCheckpointOnFailedStoring(
+                            completedCheckpoint, executor);
+                }
 
                 sendAbortedMessages(
                         pendingCheckpoint.getCheckpointPlan().getTasksToCommitTo(),

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

@@ -20,6 +20,7 @@
 
 import org.apache.flink.api.common.JobStatus;
 import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.runtime.persistence.PossibleInconsistentStateException;
 import org.apache.flink.runtime.persistence.ResourceVersion;
 import org.apache.flink.runtime.persistence.StateHandleStore;
 import org.apache.flink.runtime.state.RetrievableStateHandle;
@@ -212,6 +213,9 @@ public void recover() throws Exception {
      * older ones.
      *
      * @param checkpoint Completed checkpoint to add.
+     * @throws PossibleInconsistentStateException if adding the checkpoint failed and leaving the
+     *     system in an possibly inconsistent state, i.e. it's uncertain whether the checkpoint
+     *     metadata was fully written to the underlying systems or not.
      */
     @Override
     public void addCheckpoint(

flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorFailureTest.java

@@ -23,27 +23,36 @@
 import org.apache.flink.runtime.checkpoint.channel.InputChannelInfo;
 import org.apache.flink.runtime.checkpoint.channel.ResultSubpartitionInfo;
 import org.apache.flink.runtime.concurrent.ManuallyTriggeredScheduledExecutor;
+import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
 import org.apache.flink.runtime.executiongraph.ExecutionGraph;
 import org.apache.flink.runtime.executiongraph.ExecutionVertex;
 import org.apache.flink.runtime.jobgraph.JobVertexID;
 import org.apache.flink.runtime.jobgraph.OperatorID;
 import org.apache.flink.runtime.messages.checkpoint.AcknowledgeCheckpoint;
+import org.apache.flink.runtime.persistence.PossibleInconsistentStateException;
 import org.apache.flink.runtime.state.InputChannelStateHandle;
 import org.apache.flink.runtime.state.KeyedStateHandle;
 import org.apache.flink.runtime.state.OperatorStateHandle;
 import org.apache.flink.runtime.state.OperatorStreamStateHandle;
 import org.apache.flink.runtime.state.ResultSubpartitionStateHandle;
 import org.apache.flink.runtime.state.StreamStateHandle;
+import org.apache.flink.util.FlinkRuntimeException;
 import org.apache.flink.util.TestLogger;
 import org.apache.flink.util.function.TriConsumerWithException;
 
+import org.junit.Rule;
 import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
 
 import java.util.Collections;
 import java.util.List;
+import java.util.concurrent.Executor;
+import java.util.concurrent.atomic.AtomicInteger;
 
+import static org.hamcrest.CoreMatchers.is;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 import static org.mockito.Mockito.mock;
@@ -54,6 +63,8 @@
 /** Tests for failure of checkpoint coordinator. */
 public class CheckpointCoordinatorFailureTest extends TestLogger {
 
+    @Rule public TemporaryFolder tmpFolder = new TemporaryFolder();
+
     /**
      * Tests that a failure while storing a completed checkpoint in the completed checkpoint store
      * will properly fail the originating pending checkpoint and clean upt the completed checkpoint.
@@ -166,6 +177,81 @@ public void testFailingCompletedCheckpointStoreAdd() throws Exception {
                 .discardState();
     }
 
+    @Test
+    public void testCleanupForGenericFailure() throws Exception {
+        testStoringFailureHandling(new FlinkRuntimeException("Expected exception"), 1);
+    }
+
+    @Test
+    public void testCleanupOmissionForPossibleInconsistentStateException() throws Exception {
+        testStoringFailureHandling(new PossibleInconsistentStateException(), 0);
+    }
+
+    private void testStoringFailureHandling(Exception failure, int expectedCleanupCalls)
+            throws Exception {
+        final JobVertexID jobVertexID1 = new JobVertexID();
+
+        final ExecutionGraph graph =
+                new CheckpointCoordinatorTestingUtils.CheckpointExecutionGraphBuilder()
+                        .addJobVertex(jobVertexID1)
+                        .build();
+
+        final ExecutionVertex vertex = graph.getJobVertex(jobVertexID1).getTaskVertices()[0];
+        final ExecutionAttemptID attemptId = vertex.getCurrentExecutionAttempt().getAttemptId();
+
+        final StandaloneCheckpointIDCounter checkpointIDCounter =
+                new StandaloneCheckpointIDCounter();
+
+        final ManuallyTriggeredScheduledExecutor manuallyTriggeredScheduledExecutor =
+                new ManuallyTriggeredScheduledExecutor();
+
+        final CompletedCheckpointStore completedCheckpointStore =
+                new FailingCompletedCheckpointStore(
+                        (checkpoint, ignoredCleaner, ignoredPostCleanCallback) -> {
+                            throw failure;
+                        });
+
+        final AtomicInteger cleanupCallCount = new AtomicInteger(0);
+        final CheckpointCoordinator checkpointCoordinator =
+                new CheckpointCoordinatorBuilder()
+                        .setExecutionGraph(graph)
+                        .setCheckpointIDCounter(checkpointIDCounter)
+                        .setCheckpointsCleaner(
+                                new CheckpointsCleaner() {
+
+                                    private static final long serialVersionUID =
+                                            2029876992397573325L;
+
+                                    @Override
+                                    public void cleanCheckpointOnFailedStoring(
+                                            CompletedCheckpoint completedCheckpoint,
+                                            Executor executor) {
+                                        cleanupCallCount.incrementAndGet();
+                                        super.cleanCheckpointOnFailedStoring(
+                                                completedCheckpoint, executor);
+                                    }
+                                })
+                        .setCompletedCheckpointStore(completedCheckpointStore)
+                        .setTimer(manuallyTriggeredScheduledExecutor)
+                        .build();
+        checkpointCoordinator.triggerSavepoint(tmpFolder.newFolder().getAbsolutePath());
+        manuallyTriggeredScheduledExecutor.triggerAll();
+
+        try {
+            checkpointCoordinator.receiveAcknowledgeMessage(
+                    new AcknowledgeCheckpoint(
+                            graph.getJobID(), attemptId, checkpointIDCounter.getLast()),
+                    "unknown location");
+            fail("CheckpointException should have been thrown.");
+        } catch (CheckpointException e) {
+            assertThat(
+                    e.getCheckpointFailureReason(),
+                    is(CheckpointFailureReason.FINALIZE_CHECKPOINT_FAILURE));
+        }
+
+        assertThat(cleanupCallCount.get(), is(expectedCleanupCalls));
+    }
+
     private static final class FailingCompletedCheckpointStore implements CompletedCheckpointStore {
 
         private final TriConsumerWithException<

flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTestingUtils.java

@@ -695,6 +695,12 @@ public CheckpointCoordinatorBuilder setCoordinatorsToCheckpoint(
             return this;
         }
 
+        public CheckpointCoordinatorBuilder setCheckpointsCleaner(
+                CheckpointsCleaner checkpointsCleaner) {
+            this.checkpointsCleaner = checkpointsCleaner;
+            return this;
+        }
+
         public CheckpointCoordinatorBuilder setCheckpointIDCounter(
                 CheckpointIDCounter checkpointIDCounter) {
             this.checkpointIDCounter = checkpointIDCounter;