[FLINK-20675][checkpointing] Refactor interfaces to decline checkpoin…

…t with CheckpointException
is · Jan 28, 2021 · f421c6f · f421c6f
1 parent c7ed2e8
commit f421c6f
Show file tree

Hide file tree

Showing 28 changed files with 88 additions and 82 deletions.
diff --git a/...processing-api/src/main/java/org/apache/flink/state/api/runtime/SavepointEnvironment.java b/...processing-api/src/main/java/org/apache/flink/state/api/runtime/SavepointEnvironment.java
@@ -28,6 +28,7 @@
 import org.apache.flink.core.fs.Path;
 import org.apache.flink.runtime.accumulators.AccumulatorRegistry;
 import org.apache.flink.runtime.broadcast.BroadcastVariableManager;
+import org.apache.flink.runtime.checkpoint.CheckpointException;
 import org.apache.flink.runtime.checkpoint.CheckpointMetrics;
 import org.apache.flink.runtime.checkpoint.PrioritizedOperatorSubtaskState;
 import org.apache.flink.runtime.checkpoint.TaskStateSnapshot;
@@ -245,7 +246,7 @@ public void acknowledgeCheckpoint(
  }
 
  @Override
- public void declineCheckpoint(long checkpointId, Throwable cause) {
+ public void declineCheckpoint(long checkpointId, CheckpointException checkpointException) {
  throw new UnsupportedOperationException(ERROR_MSG);
  }
 

diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/execution/Environment.java b/flink-runtime/src/main/java/org/apache/flink/runtime/execution/Environment.java
@@ -25,6 +25,7 @@
 import org.apache.flink.core.fs.Path;
 import org.apache.flink.runtime.accumulators.AccumulatorRegistry;
 import org.apache.flink.runtime.broadcast.BroadcastVariableManager;
+import org.apache.flink.runtime.checkpoint.CheckpointException;
 import org.apache.flink.runtime.checkpoint.CheckpointMetrics;
 import org.apache.flink.runtime.checkpoint.TaskStateSnapshot;
 import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
@@ -203,9 +204,9 @@ void acknowledgeCheckpoint(
  * to successfully complete a certain checkpoint.
  *
  * @param checkpointId The ID of the declined checkpoint.
- * @param cause An optional reason why the checkpoint was declined.
+ * @param checkpointException The exception why the checkpoint was declined.
  */
- void declineCheckpoint(long checkpointId, Throwable cause);
+ void declineCheckpoint(long checkpointId, CheckpointException checkpointException);
 
  /**
  * Marks task execution failed for an external reason (a reason other than the task code itself

diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/tasks/AbstractInvokable.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/tasks/AbstractInvokable.java
@@ -20,6 +20,7 @@
 
 import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.checkpoint.CheckpointException;
 import org.apache.flink.runtime.checkpoint.CheckpointMetaData;
 import org.apache.flink.runtime.checkpoint.CheckpointMetricsBuilder;
 import org.apache.flink.runtime.checkpoint.CheckpointOptions;
@@ -253,7 +254,8 @@ public void triggerCheckpointOnBarrier(
  * @param checkpointId The ID of the checkpoint to be aborted.
  * @param cause The reason why the checkpoint was aborted during alignment
  */
- public void abortCheckpointOnBarrier(long checkpointId, Throwable cause) throws IOException {
+ public void abortCheckpointOnBarrier(long checkpointId, CheckpointException cause)
+ throws IOException {
  throw new UnsupportedOperationException(
  String.format(
  "abortCheckpointOnBarrier not supported by %s", this.getClass().getName()));

diff --git a/...ntime/src/main/java/org/apache/flink/runtime/taskexecutor/rpc/RpcCheckpointResponder.java b/...ntime/src/main/java/org/apache/flink/runtime/taskexecutor/rpc/RpcCheckpointResponder.java
@@ -64,18 +64,10 @@ public void declineCheckpoint(
  JobID jobID,
  ExecutionAttemptID executionAttemptID,
  long checkpointId,
- Throwable cause) {
+ CheckpointException checkpointException) {
 
- // TODO the passed parameter 'cause' is actually always instance of CheckpointException,
- // we should change the interfaces to narrow all declined checkpoint's throwable to
- // CheckpointException.
- Preconditions.checkArgument(
- cause instanceof CheckpointException,
- "The given cause is "
- + cause.getClass()
- + " instead of expected CheckpointException.");
  checkpointCoordinatorGateway.declineCheckpoint(
  new DeclineCheckpoint(
- jobID, executionAttemptID, checkpointId, (CheckpointException) cause));
+ jobID, executionAttemptID, checkpointId, checkpointException));
  }
 }
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/CheckpointResponder.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/CheckpointResponder.java
@@ -20,6 +20,7 @@
 
 import org.apache.flink.annotation.Internal;
 import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.checkpoint.CheckpointException;
 import org.apache.flink.runtime.checkpoint.CheckpointMetrics;
 import org.apache.flink.runtime.checkpoint.TaskStateSnapshot;
 import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
@@ -64,8 +65,11 @@ void reportCheckpointMetrics(
  * @param jobID Job ID of the running job
  * @param executionAttemptID Execution attempt ID of the running task
  * @param checkpointId The ID of the declined checkpoint
- * @param cause The optional cause why the checkpoint was declined
+ * @param checkpointException The exception why the checkpoint was declined
  */
  void declineCheckpoint(
- JobID jobID, ExecutionAttemptID executionAttemptID, long checkpointId, Throwable cause);
+ JobID jobID,
+ ExecutionAttemptID executionAttemptID,
+ long checkpointId,
+ CheckpointException checkpointException);
 }
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/RuntimeEnvironment.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/RuntimeEnvironment.java
@@ -25,6 +25,7 @@
 import org.apache.flink.core.fs.Path;
 import org.apache.flink.runtime.accumulators.AccumulatorRegistry;
 import org.apache.flink.runtime.broadcast.BroadcastVariableManager;
+import org.apache.flink.runtime.checkpoint.CheckpointException;
 import org.apache.flink.runtime.checkpoint.CheckpointMetrics;
 import org.apache.flink.runtime.checkpoint.TaskStateSnapshot;
 import org.apache.flink.runtime.execution.Environment;
@@ -292,8 +293,9 @@ public void acknowledgeCheckpoint(
  }
 
  @Override
- public void declineCheckpoint(long checkpointId, Throwable cause) {
- checkpointResponder.declineCheckpoint(jobId, executionId, checkpointId, cause);
+ public void declineCheckpoint(long checkpointId, CheckpointException checkpointException) {
+ checkpointResponder.declineCheckpoint(
+ jobId, executionId, checkpointId, checkpointException);
  }
 
  @Override

diff --git a/.../src/test/java/org/apache/flink/runtime/operators/testutils/DummyCheckpointInvokable.java b/.../src/test/java/org/apache/flink/runtime/operators/testutils/DummyCheckpointInvokable.java
@@ -18,6 +18,7 @@
 
 package org.apache.flink.runtime.operators.testutils;
 
+import org.apache.flink.runtime.checkpoint.CheckpointException;
 import org.apache.flink.runtime.checkpoint.CheckpointMetaData;
 import org.apache.flink.runtime.checkpoint.CheckpointMetricsBuilder;
 import org.apache.flink.runtime.checkpoint.CheckpointOptions;
@@ -34,5 +35,5 @@ public void triggerCheckpointOnBarrier(
  CheckpointOptions checkpointOptions,
  CheckpointMetricsBuilder checkpointMetrics) {}
 
- public void abortCheckpointOnBarrier(long checkpointId, Throwable cause) {}
+ public void abortCheckpointOnBarrier(long checkpointId, CheckpointException cause) {}
 }
diff --git a/...-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/DummyEnvironment.java b/...-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/DummyEnvironment.java
@@ -25,6 +25,7 @@
 import org.apache.flink.core.fs.Path;
 import org.apache.flink.runtime.accumulators.AccumulatorRegistry;
 import org.apache.flink.runtime.broadcast.BroadcastVariableManager;
+import org.apache.flink.runtime.checkpoint.CheckpointException;
 import org.apache.flink.runtime.checkpoint.CheckpointMetrics;
 import org.apache.flink.runtime.checkpoint.TaskStateSnapshot;
 import org.apache.flink.runtime.execution.Environment;
@@ -213,7 +214,7 @@ public void acknowledgeCheckpoint(
  TaskStateSnapshot subtaskState) {}
 
  @Override
- public void declineCheckpoint(long checkpointId, Throwable cause) {
+ public void declineCheckpoint(long checkpointId, CheckpointException cause) {
  throw new UnsupportedOperationException();
  }
 

diff --git a/...k-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/MockEnvironment.java b/...k-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/MockEnvironment.java
@@ -25,6 +25,7 @@
 import org.apache.flink.core.fs.Path;
 import org.apache.flink.runtime.accumulators.AccumulatorRegistry;
 import org.apache.flink.runtime.broadcast.BroadcastVariableManager;
+import org.apache.flink.runtime.checkpoint.CheckpointException;
 import org.apache.flink.runtime.checkpoint.CheckpointMetrics;
 import org.apache.flink.runtime.checkpoint.TaskStateSnapshot;
 import org.apache.flink.runtime.execution.Environment;
@@ -351,7 +352,7 @@ public void acknowledgeCheckpoint(
  }
 
  @Override
- public void declineCheckpoint(long checkpointId, Throwable cause) {
+ public void declineCheckpoint(long checkpointId, CheckpointException cause) {
  throw new UnsupportedOperationException(cause);
  }
 

diff --git a/...k-runtime/src/test/java/org/apache/flink/runtime/taskmanager/NoOpCheckpointResponder.java b/...k-runtime/src/test/java/org/apache/flink/runtime/taskmanager/NoOpCheckpointResponder.java
@@ -19,6 +19,7 @@
 package org.apache.flink.runtime.taskmanager;
 
 import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.checkpoint.CheckpointException;
 import org.apache.flink.runtime.checkpoint.CheckpointMetrics;
 import org.apache.flink.runtime.checkpoint.TaskStateSnapshot;
 import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
@@ -31,13 +32,13 @@ public enum NoOpCheckpointResponder implements CheckpointResponder {
  public void acknowledgeCheckpoint(
  JobID j, ExecutionAttemptID e, long i, CheckpointMetrics c, TaskStateSnapshot s) {}
 
- @Override
- public void declineCheckpoint(JobID j, ExecutionAttemptID e, long l, Throwable t) {}
-
  @Override
  public void reportCheckpointMetrics(
  JobID jobID,
  ExecutionAttemptID executionAttemptID,
  long checkpointId,
  CheckpointMetrics checkpointMetrics) {}
+
+ @Override
+ public void declineCheckpoint(JobID j, ExecutionAttemptID e, long l, CheckpointException c) {}
 }
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskAsyncCallTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskAsyncCallTest.java
@@ -23,6 +23,7 @@
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.core.testutils.OneShotLatch;
 import org.apache.flink.runtime.broadcast.BroadcastVariableManager;
+import org.apache.flink.runtime.checkpoint.CheckpointException;
 import org.apache.flink.runtime.checkpoint.CheckpointMetaData;
 import org.apache.flink.runtime.checkpoint.CheckpointMetricsBuilder;
 import org.apache.flink.runtime.checkpoint.CheckpointOptions;
@@ -290,7 +291,7 @@ public void triggerCheckpointOnBarrier(
  }
 
  @Override
- public void abortCheckpointOnBarrier(long checkpointId, Throwable cause) {
+ public void abortCheckpointOnBarrier(long checkpointId, CheckpointException cause) {
  throw new UnsupportedOperationException("Should not be called");
  }
 

diff --git a/...k-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TestCheckpointResponder.java b/...k-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TestCheckpointResponder.java
@@ -20,6 +20,7 @@
 
 import org.apache.flink.api.common.JobID;
 import org.apache.flink.core.testutils.OneShotLatch;
+import org.apache.flink.runtime.checkpoint.CheckpointException;
 import org.apache.flink.runtime.checkpoint.CheckpointMetrics;
 import org.apache.flink.runtime.checkpoint.TaskStateSnapshot;
 import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
@@ -72,10 +73,10 @@ public void declineCheckpoint(
  JobID jobID,
  ExecutionAttemptID executionAttemptID,
  long checkpointId,
- Throwable cause) {
+ CheckpointException checkpointException) {
 
  DeclineReport declineReport =
- new DeclineReport(jobID, executionAttemptID, checkpointId, cause);
+ new DeclineReport(jobID, executionAttemptID, checkpointId, checkpointException);
 
  declineReports.add(declineReport);
 
@@ -137,19 +138,19 @@ public TaskStateSnapshot getSubtaskState() {
 
  public static class DeclineReport extends AbstractReport {
 
- public final Throwable cause;
+ public final CheckpointException cause;
 
  public DeclineReport(
  JobID jobID,
  ExecutionAttemptID executionAttemptID,
  long checkpointId,
- Throwable cause) {
+ CheckpointException cause) {
 
  super(jobID, executionAttemptID, checkpointId);
  this.cause = cause;
  }
 
- public Throwable getCause() {
+ public CheckpointException getCause() {
  return cause;
  }
  }

diff --git a/...ksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBAsyncSnapshotTest.java b/...ksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBAsyncSnapshotTest.java
@@ -28,6 +28,7 @@
 import org.apache.flink.core.fs.CloseableRegistry;
 import org.apache.flink.core.testutils.OneShotLatch;
 import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
+import org.apache.flink.runtime.checkpoint.CheckpointException;
 import org.apache.flink.runtime.checkpoint.CheckpointMetaData;
 import org.apache.flink.runtime.checkpoint.CheckpointMetrics;
 import org.apache.flink.runtime.checkpoint.CheckpointOptions;
@@ -197,7 +198,7 @@ public void declineCheckpoint(
  JobID jobID,
  ExecutionAttemptID executionAttemptID,
  long checkpointId,
- Throwable cause) {}
+ CheckpointException checkpointException) {}
  };
 
  JobID jobID = new JobID();

diff --git a/...-java/src/main/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTask.java b/...-java/src/main/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTask.java
@@ -18,6 +18,7 @@
 package org.apache.flink.streaming.runtime.tasks;
 
 import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.checkpoint.CheckpointException;
 import org.apache.flink.runtime.checkpoint.CheckpointMetaData;
 import org.apache.flink.runtime.checkpoint.CheckpointMetricsBuilder;
 import org.apache.flink.runtime.checkpoint.CheckpointOptions;
@@ -250,7 +251,8 @@ public void triggerCheckpointOnBarrier(
  }
 
  @Override
- public void abortCheckpointOnBarrier(long checkpointId, Throwable cause) throws IOException {
+ public void abortCheckpointOnBarrier(long checkpointId, CheckpointException cause)
+ throws IOException {
  CompletableFuture<Boolean> resultFuture =
  pendingCheckpointCompletedFutures.remove(checkpointId);
  if (resultFuture != null) {

diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
@@ -986,7 +986,8 @@ public void triggerCheckpointOnBarrier(
  }
 
  @Override
- public void abortCheckpointOnBarrier(long checkpointId, Throwable cause) throws IOException {
+ public void abortCheckpointOnBarrier(long checkpointId, CheckpointException cause)
+ throws IOException {
  subtaskCheckpointCoordinator.abortCheckpointOnBarrier(checkpointId, cause, operatorChain);
  }
 

diff --git a/.../src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinator.java b/.../src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinator.java
@@ -18,6 +18,7 @@
 package org.apache.flink.streaming.runtime.tasks;
 
 import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.checkpoint.CheckpointException;
 import org.apache.flink.runtime.checkpoint.CheckpointMetaData;
 import org.apache.flink.runtime.checkpoint.CheckpointMetricsBuilder;
 import org.apache.flink.runtime.checkpoint.CheckpointOptions;
@@ -50,7 +51,7 @@ public interface SubtaskCheckpointCoordinator extends Closeable {
  CheckpointStorageWorkerView getCheckpointStorage();
 
  void abortCheckpointOnBarrier(
- long checkpointId, Throwable cause, OperatorChain<?, ?> operatorChain)
+ long checkpointId, CheckpointException cause, OperatorChain<?, ?> operatorChain)
  throws IOException;
 
  /** Must be called after {@link #initCheckpoint(long, CheckpointOptions)}. */

diff --git a/.../main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java b/.../main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java
@@ -196,7 +196,7 @@ private static ChannelStateWriter openChannelStateWriter(
 
  @Override
  public void abortCheckpointOnBarrier(
- long checkpointId, Throwable cause, OperatorChain<?, ?> operatorChain)
+ long checkpointId, CheckpointException cause, OperatorChain<?, ?> operatorChain)
  throws IOException {
  LOG.debug("Aborting checkpoint via cancel-barrier {} for task {}", checkpointId, taskName);
  lastCheckpointId = Math.max(lastCheckpointId, checkpointId);

diff --git a/...java/org/apache/flink/streaming/runtime/io/checkpointing/CheckpointSequenceValidator.java b/...java/org/apache/flink/streaming/runtime/io/checkpointing/CheckpointSequenceValidator.java
@@ -18,6 +18,7 @@
 
 package org.apache.flink.streaming.runtime.io.checkpointing;
 
+import org.apache.flink.runtime.checkpoint.CheckpointException;
 import org.apache.flink.runtime.checkpoint.CheckpointMetaData;
 import org.apache.flink.runtime.checkpoint.CheckpointMetricsBuilder;
 import org.apache.flink.runtime.checkpoint.CheckpointOptions;
@@ -79,7 +80,7 @@ public void triggerCheckpointOnBarrier(
  }
 
  @Override
- public void abortCheckpointOnBarrier(long checkpointId, Throwable cause) {
+ public void abortCheckpointOnBarrier(long checkpointId, CheckpointException cause) {
  assertTrue(
  "Unexpected abortCheckpointOnBarrier(" + checkpointId + ")",
  i < checkpointIDs.length);

diff --git a/.../apache/flink/streaming/runtime/io/checkpointing/UnalignedControllerCancellationTest.java b/.../apache/flink/streaming/runtime/io/checkpointing/UnalignedControllerCancellationTest.java
@@ -17,6 +17,7 @@
 
 package org.apache.flink.streaming.runtime.io.checkpointing;
 
+import org.apache.flink.runtime.checkpoint.CheckpointException;
 import org.apache.flink.runtime.checkpoint.CheckpointMetaData;
 import org.apache.flink.runtime.checkpoint.CheckpointMetricsBuilder;
 import org.apache.flink.runtime.checkpoint.CheckpointOptions;
@@ -141,7 +142,7 @@ public void triggerCheckpointOnBarrier(
  }
 
  @Override
- public void abortCheckpointOnBarrier(long checkpointId, Throwable cause) {
+ public void abortCheckpointOnBarrier(long checkpointId, CheckpointException cause) {
  checkpointAborted = true;
  }
  }

diff --git a/...est/java/org/apache/flink/streaming/runtime/io/checkpointing/UnalignedControllerTest.java b/...est/java/org/apache/flink/streaming/runtime/io/checkpointing/UnalignedControllerTest.java
@@ -17,6 +17,7 @@
 
 package org.apache.flink.streaming.runtime.io.checkpointing;
 
+import org.apache.flink.runtime.checkpoint.CheckpointException;
 import org.apache.flink.runtime.checkpoint.CheckpointMetaData;
 import org.apache.flink.runtime.checkpoint.CheckpointMetricsBuilder;
 import org.apache.flink.runtime.checkpoint.CheckpointOptions;
@@ -968,7 +969,7 @@ public ValidatingCheckpointHandler(long nextExpectedCheckpointId) {
  }
 
  @Override
- public void abortCheckpointOnBarrier(long checkpointId, Throwable cause) {
+ public void abortCheckpointOnBarrier(long checkpointId, CheckpointException cause) {
  super.abortCheckpointOnBarrier(checkpointId, cause);
  nextExpectedCheckpointId = -1;
  }
@@ -997,7 +998,7 @@ public void init() {}
  protected void processInput(MailboxDefaultAction.Controller controller) {}
 
  @Override
- public void abortCheckpointOnBarrier(long checkpointId, Throwable cause)
+ public void abortCheckpointOnBarrier(long checkpointId, CheckpointException cause)
  throws IOException {
  abortedCheckpointId = checkpointId;
  }

diff --git a/...java/org/apache/flink/streaming/runtime/io/checkpointing/ValidatingCheckpointHandler.java b/...java/org/apache/flink/streaming/runtime/io/checkpointing/ValidatingCheckpointHandler.java
@@ -127,9 +127,9 @@ public void triggerCheckpointOnBarrier(
  }
 
  @Override
- public void abortCheckpointOnBarrier(long checkpointId, Throwable cause) {
+ public void abortCheckpointOnBarrier(long checkpointId, CheckpointException cause) {
  lastCanceledCheckpointId = checkpointId;
- failureReason = ((CheckpointException) cause).getCheckpointFailureReason();
+ failureReason = cause.getCheckpointFailureReason();
  abortedCheckpointCounter++;
  }