Move from CHECKPOINTING_INTERVAL to enableCheckpointing

apache · Sep 12, 2024 · 221e8bc · 221e8bc
1 parent 05815e8
commit 221e8bc
Show file tree

Hide file tree

Showing 2 changed files with 2 additions and 4 deletions.
diff --git a/...19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestLockRemover.java b/...19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestLockRemover.java
@@ -24,7 +24,6 @@
 import static org.assertj.core.api.Assertions.assertThat;
 
 import java.io.File;
-import java.time.Duration;
 import java.util.Collection;
 import org.apache.flink.api.common.functions.FlatMapFunction;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
@@ -100,8 +99,8 @@ void testInSink() throws Exception {
     Configuration config = new Configuration();
     config.set(CheckpointingOptions.CHECKPOINT_STORAGE, "filesystem");
     config.set(CheckpointingOptions.CHECKPOINTS_DIRECTORY, "file://" + checkpointDir.getPath());
-    config.set(CheckpointingOptions.CHECKPOINTING_INTERVAL, Duration.ofMillis(10));
     StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(config);
+    env.enableCheckpointing(10);
     ManualSource<TaskResult> source = new ManualSource<>(env, TypeInformation.of(TaskResult.class));
     source.dataStream().global().sinkTo(new SinkTest()).name(sinkName).setParallelism(1);
 

diff --git a/...20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestLockRemover.java b/...20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestLockRemover.java
@@ -24,7 +24,6 @@
 import static org.assertj.core.api.Assertions.assertThat;
 
 import java.io.File;
-import java.time.Duration;
 import java.util.Collection;
 import org.apache.flink.api.common.functions.FlatMapFunction;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
@@ -100,8 +99,8 @@ void testInSink() throws Exception {
     Configuration config = new Configuration();
     config.set(CheckpointingOptions.CHECKPOINT_STORAGE, "filesystem");
     config.set(CheckpointingOptions.CHECKPOINTS_DIRECTORY, "file://" + checkpointDir.getPath());
-    config.set(CheckpointingOptions.CHECKPOINTING_INTERVAL, Duration.ofMillis(10));
     StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(config);
+    env.enableCheckpointing(10);
     ManualSource<TaskResult> source = new ManualSource<>(env, TypeInformation.of(TaskResult.class));
     source.dataStream().global().sinkTo(new SinkTest()).name(sinkName).setParallelism(1);