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

Author: XComp

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

@@ -38,6 +38,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.CheckpointStorageCoordinatorView;
 import org.apache.flink.runtime.state.CheckpointStorageLocation;
 import org.apache.flink.runtime.state.CompletedCheckpointStorageLocation;
@@ -1210,8 +1211,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(checkpointId, pendingCheckpoint.getCheckpointTimestamp());
                 throw new CheckpointException(

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

@@ -28,21 +28,29 @@
 import org.apache.flink.runtime.executiongraph.ExecutionVertex;
 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.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;
@@ -53,6 +61,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.
@@ -73,7 +83,10 @@ public void testFailingCompletedCheckpointStoreAdd() throws Exception {
                 new CheckpointCoordinatorBuilder()
                         .setJobId(jid)
                         .setTasks(new ExecutionVertex[] {vertex})
-                        .setCompletedCheckpointStore(new FailingCompletedCheckpointStore())
+                        .setCompletedCheckpointStore(
+                                new FailingCompletedCheckpointStore(
+                                        new Exception(
+                                                "The failing completed checkpoint store failed again... :-(")))
                         .setTimer(manuallyTriggeredScheduledExecutor)
                         .build();
 
@@ -158,8 +171,83 @@ 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 JobID jobId = new JobID();
+
+        final ExecutionAttemptID executionAttemptId = new ExecutionAttemptID();
+        final ExecutionVertex vertex =
+                CheckpointCoordinatorTestingUtils.mockExecutionVertex(executionAttemptId);
+
+        final StandaloneCheckpointIDCounter checkpointIDCounter =
+                new StandaloneCheckpointIDCounter();
+
+        final ManuallyTriggeredScheduledExecutor manuallyTriggeredScheduledExecutor =
+                new ManuallyTriggeredScheduledExecutor();
+
+        final CompletedCheckpointStore completedCheckpointStore =
+                new FailingCompletedCheckpointStore(failure);
+
+        final AtomicInteger cleanupCallCount = new AtomicInteger(0);
+        final CheckpointCoordinator checkpointCoordinator =
+                new CheckpointCoordinatorBuilder()
+                        .setJobId(jobId)
+                        .setTasks(new ExecutionVertex[] {vertex})
+                        .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(
+                            jobId, executionAttemptId, 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 Exception addCheckpointFailure;
+
+        public FailingCompletedCheckpointStore(Exception addCheckpointFailure) {
+            this.addCheckpointFailure = addCheckpointFailure;
+        }
+
         @Override
         public void recover() throws Exception {
             throw new UnsupportedOperationException("Not implemented.");
@@ -171,7 +259,7 @@ public void addCheckpoint(
                 CheckpointsCleaner checkpointsCleaner,
                 Runnable postCleanup)
                 throws Exception {
-            throw new Exception("The failing completed checkpoint store failed again... :-(");
+            throw addCheckpointFailure;
         }
 
         @Override

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

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