[FLINK-22290][checkpointing] Use duration for alignment timeout.

is · Apr 16, 2021 · 52b400f · 52b400f
1 parent f6e08cd
commit 52b400f
Show file tree

Hide file tree

Showing 9 changed files with 22 additions and 20 deletions.
diff --git a/...aming-java/src/main/java/org/apache/flink/streaming/api/environment/CheckpointConfig.java b/...aming-java/src/main/java/org/apache/flink/streaming/api/environment/CheckpointConfig.java
@@ -36,6 +36,7 @@
 import javax.annotation.Nullable;
 
 import java.net.URI;
+import java.time.Duration;
 
 import static java.util.Objects.requireNonNull;
 import static org.apache.flink.runtime.checkpoint.CheckpointFailureManager.UNLIMITED_TOLERABLE_FAILURE_NUMBER;
@@ -90,8 +91,8 @@ public class CheckpointConfig implements java.io.Serializable {
  /** Flag to enable unaligned checkpoints. */
  private boolean unalignedCheckpointsEnabled;
 
- private long alignmentTimeout =
- ExecutionCheckpointingOptions.ALIGNMENT_TIMEOUT.defaultValue().toMillis();
+ private Duration alignmentTimeout =
+ ExecutionCheckpointingOptions.ALIGNMENT_TIMEOUT.defaultValue();
 
  /** Flag to enable approximate local recovery. */
  private boolean approximateLocalRecovery;
@@ -539,16 +540,16 @@ public boolean isUnalignedCheckpointsEnabled() {
  * checkpoint.
  */
  @PublicEvolving
- public void setAlignmentTimeout(long alignmentTimeout) {
+ public void setAlignmentTimeout(Duration alignmentTimeout) {
  this.alignmentTimeout = alignmentTimeout;
  }
 
  /**
- * @return value of alignment timeout, as configured via {@link #setAlignmentTimeout(long)} or
- * {@link ExecutionCheckpointingOptions#ALIGNMENT_TIMEOUT}.
+ * @return value of alignment timeout, as configured via {@link #setAlignmentTimeout(Duration)}
+ * or {@link ExecutionCheckpointingOptions#ALIGNMENT_TIMEOUT}.
  */
  @PublicEvolving
- public long getAlignmentTimeout() {
+ public Duration getAlignmentTimeout() {
  return alignmentTimeout;
  }
 
@@ -750,7 +751,7 @@ public void configure(ReadableConfig configuration) {
  .ifPresent(this::enableUnalignedCheckpoints);
  configuration
  .getOptional(ExecutionCheckpointingOptions.ALIGNMENT_TIMEOUT)
- .ifPresent(timeout -> setAlignmentTimeout(timeout.toMillis()));
+ .ifPresent(this::setAlignmentTimeout);
  configuration
  .getOptional(ExecutionCheckpointingOptions.FORCE_UNALIGNED)
  .ifPresent(this::setForceUnalignedCheckpoints);

diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamConfig.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamConfig.java
@@ -46,6 +46,7 @@
 
 import java.io.IOException;
 import java.io.Serializable;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
@@ -459,14 +460,12 @@ public boolean isExactlyOnceCheckpointMode() {
  return getCheckpointMode() == CheckpointingMode.EXACTLY_ONCE;
  }
 
- public long getAlignmentTimeout() {
- return config.getLong(
- ExecutionCheckpointingOptions.ALIGNMENT_TIMEOUT.key(),
- ExecutionCheckpointingOptions.ALIGNMENT_TIMEOUT.defaultValue().toMillis());
+ public Duration getAlignmentTimeout() {
+ return config.get(ExecutionCheckpointingOptions.ALIGNMENT_TIMEOUT);
  }
 
- public void setAlignmentTimeout(long alignmentTimeout) {
- config.setLong(ExecutionCheckpointingOptions.ALIGNMENT_TIMEOUT.key(), alignmentTimeout);
+ public void setAlignmentTimeout(Duration alignmentTimeout) {
+ config.set(ExecutionCheckpointingOptions.ALIGNMENT_TIMEOUT, alignmentTimeout);
  }
 
  public void setOutEdgesInOrder(List<StreamEdge> outEdgeList) {

diff --git a/...g-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java b/...g-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
@@ -1313,7 +1313,7 @@ private void configureCheckpointing() {
  .setTolerableCheckpointFailureNumber(
  cfg.getTolerableCheckpointFailureNumber())
  .setUnalignedCheckpointsEnabled(cfg.isUnalignedCheckpointsEnabled())
- .setAlignmentTimeout(cfg.getAlignmentTimeout())
+ .setAlignmentTimeout(cfg.getAlignmentTimeout().toMillis())
  .build(),
  serializedStateBackend,
  serializedCheckpointStorage,

diff --git a/...java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceOperatorStreamTask.java b/...java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceOperatorStreamTask.java
@@ -144,7 +144,7 @@ private void triggerCheckpointForExternallyInducedSource(long checkpointId) {
  CheckpointStorageLocationReference.getDefault(),
  configuration.isExactlyOnceCheckpointMode(),
  configuration.isUnalignedCheckpointsEnabled(),
- configuration.getAlignmentTimeout());
+ configuration.getAlignmentTimeout().toMillis());
  final long timestamp = System.currentTimeMillis();
 
  final CheckpointMetaData checkpointMetaData =

diff --git a/...reaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTask.java b/...reaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTask.java
@@ -109,7 +109,7 @@ public void triggerCheckpoint(long checkpointId) throws FlinkException {
  CheckpointStorageLocationReference.getDefault(),
  configuration.isExactlyOnceCheckpointMode(),
  configuration.isUnalignedCheckpointsEnabled(),
- configuration.getAlignmentTimeout());
+ configuration.getAlignmentTimeout().toMillis());
  final long timestamp = System.currentTimeMillis();
 
  final CheckpointMetaData checkpointMetaData =

diff --git a/...flink/streaming/runtime/tasks/MultipleInputStreamTaskChainedSourcesCheckpointingTest.java b/...flink/streaming/runtime/tasks/MultipleInputStreamTaskChainedSourcesCheckpointingTest.java
@@ -245,7 +245,7 @@ private CheckpointBarrier createBarrier(StreamTaskMailboxTestHarness<String> tes
  CheckpointStorageLocationReference.getDefault(),
  config.isExactlyOnceCheckpointMode(),
  config.isUnalignedCheckpointsEnabled(),
- config.getAlignmentTimeout());
+ config.getAlignmentTimeout().toMillis());
 
  return new CheckpointBarrier(
  metaData.getCheckpointId(), metaData.getTimestamp(), checkpointOptions);

diff --git a/...a/src/test/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTaskTest.java b/...a/src/test/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTaskTest.java
@@ -77,6 +77,7 @@
 import org.junit.Test;
 
 import java.io.Serializable;
+import java.time.Duration;
 import java.util.ArrayDeque;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -993,7 +994,7 @@ static StreamTaskMailboxTestHarness<String> buildTestHarness(boolean unaligned)
  MultipleInputStreamTask::new, BasicTypeInfo.STRING_TYPE_INFO)
  .modifyExecutionConfig(config -> config.enableObjectReuse())
  .modifyStreamConfig(config -> config.setUnalignedCheckpointsEnabled(unaligned))
- .modifyStreamConfig(config -> config.setAlignmentTimeout(0))
+ .modifyStreamConfig(config -> config.setAlignmentTimeout(Duration.ZERO))
  .addInput(BasicTypeInfo.STRING_TYPE_INFO)
  .addSourceInput(
  new SourceOperatorFactory<>(

diff --git a/...test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointCompatibilityITCase.java b/...test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointCompatibilityITCase.java
@@ -45,6 +45,7 @@
 import java.io.File;
 import java.io.FilenameFilter;
 import java.io.IOException;
+import java.time.Duration;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Comparator;
@@ -206,7 +207,7 @@ private StreamExecutionEnvironment env(boolean isAligned, int checkpointingInter
  env.setParallelism(PARALLELISM);
  env.setRestartStrategy(new RestartStrategies.NoRestartStrategyConfiguration());
  env.getCheckpointConfig().enableUnalignedCheckpoints(!isAligned);
- env.getCheckpointConfig().setAlignmentTimeout(0);
+ env.getCheckpointConfig().setAlignmentTimeout(Duration.ZERO);
  env.getCheckpointConfig().enableExternalizedCheckpoints(RETAIN_ON_CANCELLATION);
  if (checkpointingInterval > 0) {
  env.enableCheckpointing(checkpointingInterval);

diff --git a/...-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointTestBase.java b/...-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointTestBase.java
@@ -744,7 +744,7 @@ public UnalignedSettings setTolerableCheckpointFailures(int tolerableCheckpointF
 
  public void configure(StreamExecutionEnvironment env) {
  env.enableCheckpointing(Math.max(100L, parallelism * 50L));
- env.getCheckpointConfig().setAlignmentTimeout(alignmentTimeout);
+ env.getCheckpointConfig().setAlignmentTimeout(Duration.ofMillis(alignmentTimeout));
  env.getCheckpointConfig().setCheckpointTimeout(checkpointTimeout.toMillis());
  env.getCheckpointConfig()
  .setTolerableCheckpointFailureNumber(tolerableCheckpointFailures);