[FLINK-34516][checkpoint] Use new CheckpointingMode in flink-core everywhere

Author: Zakelly

flink-clients/src/test/java/org/apache/flink/client/program/StreamContextEnvironmentTest.java

@@ -25,10 +25,10 @@
 import org.apache.flink.configuration.ExecutionOptions;
 import org.apache.flink.configuration.PipelineOptions;
 import org.apache.flink.configuration.StateRecoveryOptions;
+import org.apache.flink.core.execution.CheckpointingMode;
 import org.apache.flink.core.execution.PipelineExecutorFactory;
 import org.apache.flink.core.execution.PipelineExecutorServiceLoader;
 import org.apache.flink.runtime.state.storage.JobManagerCheckpointStorage;
-import org.apache.flink.streaming.api.CheckpointingMode;
 import org.apache.flink.streaming.api.environment.CheckpointConfig;
 import org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/sink/FileSinkCompactionSwitchITCase.java

@@ -34,6 +34,7 @@
 import org.apache.flink.connector.file.sink.utils.IntegerFileSinkTestDataUtils.IntEncoder;
 import org.apache.flink.connector.file.sink.utils.IntegerFileSinkTestDataUtils.ModuloBucketAssigner;
 import org.apache.flink.connector.file.sink.utils.PartSizeAndCheckpointRollingPolicy;
+import org.apache.flink.core.execution.CheckpointingMode;
 import org.apache.flink.core.execution.SavepointFormatType;
 import org.apache.flink.core.fs.Path;
 import org.apache.flink.runtime.client.JobExecutionException;
@@ -48,7 +49,6 @@
 import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
 import org.apache.flink.runtime.state.storage.FileSystemCheckpointStorage;
 import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
-import org.apache.flink.streaming.api.CheckpointingMode;
 import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/sink/StreamingExecutionFileSinkITCase.java

@@ -25,11 +25,11 @@
 import org.apache.flink.api.common.time.Time;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.ExecutionOptions;
+import org.apache.flink.core.execution.CheckpointingMode;
 import org.apache.flink.runtime.jobgraph.JobGraph;
 import org.apache.flink.runtime.state.CheckpointListener;
 import org.apache.flink.runtime.state.FunctionInitializationContext;
 import org.apache.flink.runtime.state.FunctionSnapshotContext;
-import org.apache.flink.streaming.api.CheckpointingMode;
 import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
 import org.apache.flink.streaming.api.datastream.DataStreamSink;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/sink/writer/FileSinkMigrationITCase.java

@@ -25,6 +25,7 @@
 import org.apache.flink.api.common.state.ListStateDescriptor;
 import org.apache.flink.connector.file.sink.FileSink;
 import org.apache.flink.connector.file.sink.utils.IntegerFileSinkTestDataUtils;
+import org.apache.flink.core.execution.CheckpointingMode;
 import org.apache.flink.core.execution.SavepointFormatType;
 import org.apache.flink.core.fs.Path;
 import org.apache.flink.runtime.jobgraph.JobGraph;
@@ -33,7 +34,6 @@
 import org.apache.flink.runtime.minicluster.MiniClusterConfiguration;
 import org.apache.flink.runtime.state.FunctionInitializationContext;
 import org.apache.flink.runtime.state.FunctionSnapshotContext;
-import org.apache.flink.streaming.api.CheckpointingMode;
 import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink;

flink-end-to-end-tests/flink-datastream-allround-test/src/main/java/org/apache/flink/streaming/tests/DataStreamAllroundTestJobFactory.java

@@ -33,8 +33,8 @@
 import org.apache.flink.configuration.ConfigOption;
 import org.apache.flink.configuration.ConfigOptions;
 import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
+import org.apache.flink.core.execution.CheckpointingMode;
 import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
-import org.apache.flink.streaming.api.CheckpointingMode;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.datastream.KeyedStream;
 import org.apache.flink.streaming.api.datastream.WindowedStream;

flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroBulkFormatITCase.java

@@ -32,8 +32,8 @@
 import org.apache.flink.connector.testframe.junit.annotations.TestEnv;
 import org.apache.flink.connector.testframe.junit.annotations.TestSemantics;
 import org.apache.flink.connector.testframe.testsuites.SourceTestSuiteBase;
+import org.apache.flink.core.execution.CheckpointingMode;
 import org.apache.flink.formats.avro.typeutils.AvroSchemaConverter;
-import org.apache.flink.streaming.api.CheckpointingMode;
 import org.apache.flink.table.data.GenericRowData;
 import org.apache.flink.table.data.RowData;
 import org.apache.flink.table.data.StringData;

flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/BootstrapTransformation.java

@@ -27,6 +27,7 @@
 import org.apache.flink.api.java.functions.KeySelector;
 import org.apache.flink.api.java.operators.MapPartitionOperator;
 import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.execution.CheckpointingMode;
 import org.apache.flink.core.fs.Path;
 import org.apache.flink.core.memory.ManagedMemoryUseCase;
 import org.apache.flink.runtime.checkpoint.OperatorState;
@@ -39,7 +40,6 @@
 import org.apache.flink.state.api.output.operators.BroadcastStateBootstrapOperator;
 import org.apache.flink.state.api.output.partitioner.HashSelector;
 import org.apache.flink.state.api.output.partitioner.KeyGroupRangePartitioner;
-import org.apache.flink.streaming.api.CheckpointingMode;
 import org.apache.flink.streaming.api.graph.StreamConfig;
 import org.apache.flink.streaming.api.operators.StreamOperator;
 

flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/StateBootstrapTransformation.java

@@ -25,6 +25,7 @@
 import org.apache.flink.api.common.typeutils.TypeSerializer;
 import org.apache.flink.api.java.functions.KeySelector;
 import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.execution.CheckpointingMode;
 import org.apache.flink.core.fs.Path;
 import org.apache.flink.core.memory.ManagedMemoryUseCase;
 import org.apache.flink.runtime.checkpoint.OperatorState;
@@ -36,7 +37,6 @@
 import org.apache.flink.state.api.output.operators.BroadcastStateBootstrapOperator;
 import org.apache.flink.state.api.output.operators.GroupReduceOperator;
 import org.apache.flink.state.api.runtime.MutableConfig;
-import org.apache.flink.streaming.api.CheckpointingMode;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
 import org.apache.flink.streaming.api.graph.StreamConfig;

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/CheckpointConfig.java

@@ -29,12 +29,12 @@
 import org.apache.flink.configuration.ReadableConfig;
 import org.apache.flink.configuration.StateRecoveryOptions;
 import org.apache.flink.configuration.description.InlineElement;
+import org.apache.flink.core.execution.CheckpointingMode;
 import org.apache.flink.core.fs.Path;
 import org.apache.flink.runtime.jobgraph.tasks.CheckpointCoordinatorConfiguration;
 import org.apache.flink.runtime.state.CheckpointStorage;
 import org.apache.flink.runtime.state.StateBackend;
 import org.apache.flink.runtime.state.storage.FileSystemCheckpointStorage;
-import org.apache.flink.streaming.api.CheckpointingMode;
 import org.apache.flink.util.Preconditions;
 
 import org.slf4j.Logger;
@@ -72,9 +72,9 @@ public class CheckpointConfig implements java.io.Serializable {
      * The default checkpoint mode: exactly once.
      *
      * @deprecated This field is no longer used. Please use {@link
-     *     ExecutionCheckpointingOptions.CHECKPOINTING_MODE} instead.
+     *     ExecutionCheckpointingOptions.CHECKPOINTING_CONSISTENCY_MODE} instead.
      */
-    public static final CheckpointingMode DEFAULT_MODE =
+    public static final org.apache.flink.streaming.api.CheckpointingMode DEFAULT_MODE =
             ExecutionCheckpointingOptions.CHECKPOINTING_MODE.defaultValue();
 
     /**
@@ -177,41 +177,42 @@ public boolean isCheckpointingEnabled() {
      * Gets the checkpointing mode (exactly-once vs. at-least-once).
      *
      * @return The checkpointing mode.
-     * @deprecated Use {@link #getCheckpointMode} instead.
+     * @deprecated Use {@link #getConsistencyMode} instead.
      */
     @Deprecated
-    public CheckpointingMode getCheckpointingMode() {
+    public org.apache.flink.streaming.api.CheckpointingMode getCheckpointingMode() {
         return configuration.get(ExecutionCheckpointingOptions.CHECKPOINTING_MODE);
     }
 
     /**
      * Sets the checkpointing mode (exactly-once vs. at-least-once).
      *
      * @param checkpointingMode The checkpointing mode.
-     * @deprecated Use {@link #setCheckpointMode} instead.
+     * @deprecated Use {@link #setConsistencyMode} instead.
      */
     @Deprecated
-    public void setCheckpointingMode(CheckpointingMode checkpointingMode) {
+    public void setCheckpointingMode(
+            org.apache.flink.streaming.api.CheckpointingMode checkpointingMode) {
         configuration.set(ExecutionCheckpointingOptions.CHECKPOINTING_MODE, checkpointingMode);
     }
 
     /**
-     * Gets the checkpointing mode (exactly-once vs. at-least-once).
+     * Gets the checkpointing consistency mode (exactly-once vs. at-least-once).
      *
      * @return The checkpointing mode.
      */
-    public org.apache.flink.core.execution.CheckpointingMode getCheckpointMode() {
-        return configuration.get(ExecutionCheckpointingOptions.CHECKPOINTING_MODE_V2);
+    public CheckpointingMode getConsistencyMode() {
+        return configuration.get(ExecutionCheckpointingOptions.CHECKPOINTING_CONSISTENCY_MODE);
     }
 
     /**
-     * Sets the checkpointing mode (exactly-once vs. at-least-once).
+     * Sets the checkpointing consistency mode (exactly-once vs. at-least-once).
      *
      * @param checkpointingMode The checkpointing mode.
      */
-    public void setCheckpointMode(
-            org.apache.flink.core.execution.CheckpointingMode checkpointingMode) {
-        configuration.set(ExecutionCheckpointingOptions.CHECKPOINTING_MODE_V2, checkpointingMode);
+    public void setConsistencyMode(CheckpointingMode checkpointingMode) {
+        configuration.set(
+                ExecutionCheckpointingOptions.CHECKPOINTING_CONSISTENCY_MODE, checkpointingMode);
     }
 
     /**
@@ -598,7 +599,8 @@ public boolean isExternalizedCheckpointsEnabled() {
      * embedded into the stream of data anymore.
      *
      * <p>Unaligned checkpoints can only be enabled if {@link
-     * ExecutionCheckpointingOptions#CHECKPOINTING_MODE} is {@link CheckpointingMode#EXACTLY_ONCE}.
+     * ExecutionCheckpointingOptions#CHECKPOINTING_CONSISTENCY_MODE} is {@link
+     * CheckpointingMode#EXACTLY_ONCE}.
      *
      * @param enabled Flag to indicate whether unaligned are enabled.
      */
@@ -616,7 +618,8 @@ public void enableUnalignedCheckpoints(boolean enabled) {
      * embedded into the stream of data anymore.
      *
      * <p>Unaligned checkpoints can only be enabled if {@link
-     * ExecutionCheckpointingOptions#CHECKPOINTING_MODE} is {@link CheckpointingMode#EXACTLY_ONCE}.
+     * ExecutionCheckpointingOptions#CHECKPOINTING_CONSISTENCY_MODE} is {@link
+     * CheckpointingMode#EXACTLY_ONCE}.
      */
     @PublicEvolving
     public void enableUnalignedCheckpoints() {
@@ -986,7 +989,7 @@ public InlineElement getDescription() {
 
     /**
      * Sets all relevant options contained in the {@link ReadableConfig} such as e.g. {@link
-     * ExecutionCheckpointingOptions#CHECKPOINTING_MODE}.
+     * ExecutionCheckpointingOptions#CHECKPOINTING_CONSISTENCY_MODE}.
      *
      * <p>It will change the value of a setting only if a corresponding option was set in the {@code
      * configuration}. If a key is not present, the current value of a field will remain untouched.
@@ -995,8 +998,8 @@ public InlineElement getDescription() {
      */
     public void configure(ReadableConfig configuration) {
         configuration
-                .getOptional(ExecutionCheckpointingOptions.CHECKPOINTING_MODE)
-                .ifPresent(this::setCheckpointingMode);
+                .getOptional(ExecutionCheckpointingOptions.CHECKPOINTING_CONSISTENCY_MODE)
+                .ifPresent(this::setConsistencyMode);
         configuration
                 .getOptional(ExecutionCheckpointingOptions.CHECKPOINTING_INTERVAL)
                 .ifPresent(i -> this.setCheckpointInterval(i.toMillis()));

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/ExecutionCheckpointingOptions.java

@@ -27,7 +27,7 @@
 import org.apache.flink.configuration.StateRecoveryOptions;
 import org.apache.flink.configuration.description.Description;
 import org.apache.flink.configuration.description.TextElement;
-import org.apache.flink.streaming.api.CheckpointingMode;
+import org.apache.flink.core.execution.CheckpointingMode;
 
 import java.time.Duration;
 
@@ -44,18 +44,20 @@ public class ExecutionCheckpointingOptions {
 
     @Deprecated
     @Documentation.ExcludeFromDocumentation("Hidden for deprecatd.")
-    public static final ConfigOption<CheckpointingMode> CHECKPOINTING_MODE =
-            ConfigOptions.key("execution.checkpointing.mode")
-                    .enumType(CheckpointingMode.class)
-                    .defaultValue(CheckpointingMode.EXACTLY_ONCE)
-                    .withDescription("The checkpointing mode (exactly-once vs. at-least-once).");
+    public static final ConfigOption<org.apache.flink.streaming.api.CheckpointingMode>
+            CHECKPOINTING_MODE =
+                    ConfigOptions.key("execution.checkpointing.mode")
+                            .enumType(org.apache.flink.streaming.api.CheckpointingMode.class)
+                            .defaultValue(
+                                    org.apache.flink.streaming.api.CheckpointingMode.EXACTLY_ONCE)
+                            .withDescription(
+                                    "The checkpointing mode (exactly-once vs. at-least-once).");
 
     public static final ConfigOption<org.apache.flink.core.execution.CheckpointingMode>
-            CHECKPOINTING_MODE_V2 =
+            CHECKPOINTING_CONSISTENCY_MODE =
                     ConfigOptions.key("execution.checkpointing.mode")
-                            .enumType(org.apache.flink.core.execution.CheckpointingMode.class)
-                            .defaultValue(
-                                    org.apache.flink.core.execution.CheckpointingMode.EXACTLY_ONCE)
+                            .enumType(CheckpointingMode.class)
+                            .defaultValue(CheckpointingMode.EXACTLY_ONCE)
                             .withDescription(
                                     "The checkpointing mode (exactly-once vs. at-least-once).");
 
@@ -201,7 +203,7 @@ public class ExecutionCheckpointingOptions {
                                     .linebreak()
                                     .text(
                                             "Unaligned checkpoints can only be enabled if %s is %s and if %s is 1",
-                                            TextElement.code(CHECKPOINTING_MODE.key()),
+                                            TextElement.code(CHECKPOINTING_CONSISTENCY_MODE.key()),
                                             TextElement.code(
                                                     CheckpointingMode.EXACTLY_ONCE.toString()),
                                             TextElement.code(MAX_CONCURRENT_CHECKPOINTS.key()))