[FLINK-34511][Checkpoint] Remove force checkpointing flag

Author: Zakelly

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

@@ -382,32 +382,6 @@ public void setMaxConcurrentCheckpoints(int maxConcurrentCheckpoints) {
                 CheckpointingOptions.MAX_CONCURRENT_CHECKPOINTS, maxConcurrentCheckpoints);
     }
 
-    /**
-     * Checks whether checkpointing is forced, despite currently non-checkpointable iteration
-     * feedback.
-     *
-     * @return True, if checkpointing is forced, false otherwise.
-     * @deprecated This will be removed once iterations properly participate in checkpointing.
-     */
-    @Deprecated
-    @PublicEvolving
-    public boolean isForceCheckpointing() {
-        return configuration.get(ExecutionCheckpointingOptions.FORCE_CHECKPOINTING);
-    }
-
-    /**
-     * Checks whether checkpointing is forced, despite currently non-checkpointable iteration
-     * feedback.
-     *
-     * @param forceCheckpointing The flag to force checkpointing.
-     * @deprecated This will be removed once iterations properly participate in checkpointing.
-     */
-    @Deprecated
-    @PublicEvolving
-    public void setForceCheckpointing(boolean forceCheckpointing) {
-        configuration.set(ExecutionCheckpointingOptions.FORCE_CHECKPOINTING, forceCheckpointing);
-    }
-
     /**
      * Checks whether unaligned checkpoints are forced, despite iteration feedback.
      *

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

@@ -338,19 +338,6 @@ public class ExecutionCheckpointingOptions {
                                                     "the important considerations"))
                                     .build());
 
-    /**
-     * Access to this option is officially only supported via {@link
-     * CheckpointConfig#setForceCheckpointing(boolean)}, but there is no good reason behind this.
-     *
-     * @deprecated This will be removed once iterations properly participate in checkpointing.
-     */
-    @Internal @Deprecated @Documentation.ExcludeFromDocumentation
-    public static final ConfigOption<Boolean> FORCE_CHECKPOINTING =
-            key("execution.checkpointing.force")
-                    .booleanType()
-                    .defaultValue(false)
-                    .withDescription("Flag to force checkpointing in iterative jobs.");
-
     /**
      * Access to this option is officially only supported via {@link
      * CheckpointConfig#enableApproximateLocalRecovery(boolean)}, but there is no good reason behind

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

@@ -559,35 +559,6 @@ public StreamExecutionEnvironment enableCheckpointing(long interval, Checkpointi
         return this;
     }
 
-    /**
-     * Enables checkpointing for the streaming job. The distributed state of the streaming dataflow
-     * will be periodically snapshotted. In case of a failure, the streaming dataflow will be
-     * restarted from the latest completed checkpoint.
-     *
-     * <p>The job draws checkpoints periodically, in the given interval. The state will be stored in
-     * the configured state backend.
-     *
-     * <p>NOTE: Checkpointing iterative streaming dataflows is not properly supported at the moment.
-     * If the "force" parameter is set to true, the system will execute the job nonetheless.
-     *
-     * @param interval Time interval between state checkpoints in millis.
-     * @param mode The checkpointing mode, selecting between "exactly once" and "at least once"
-     *     guaranteed.
-     * @param force If true checkpointing will be enabled for iterative jobs as well.
-     * @deprecated Use {@link #enableCheckpointing(long, CheckpointingMode)} instead. Forcing
-     *     checkpoints will be removed in the future.
-     */
-    @Deprecated
-    @SuppressWarnings("deprecation")
-    @PublicEvolving
-    public StreamExecutionEnvironment enableCheckpointing(
-            long interval, org.apache.flink.streaming.api.CheckpointingMode mode, boolean force) {
-        checkpointCfg.setCheckpointingMode(mode);
-        checkpointCfg.setCheckpointInterval(interval);
-        checkpointCfg.setForceCheckpointing(force);
-        return this;
-    }
-
     /**
      * Enables checkpointing for the streaming job. The distributed state of the streaming dataflow
      * will be periodically snapshotted. In case of a failure, the streaming dataflow will be
@@ -620,18 +591,6 @@ public long getCheckpointInterval() {
         return checkpointCfg.getCheckpointInterval();
     }
 
-    /**
-     * Returns whether checkpointing is force-enabled.
-     *
-     * @deprecated Forcing checkpoints will be removed in future version.
-     */
-    @Deprecated
-    @SuppressWarnings("deprecation")
-    @PublicEvolving
-    public boolean isForceCheckpointing() {
-        return checkpointCfg.isForceCheckpointing();
-    }
-
     /** Returns whether unaligned checkpoints are enabled. */
     @PublicEvolving
     public boolean isUnalignedCheckpointsEnabled() {

flink-runtime/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java

@@ -501,12 +501,6 @@ private void preValidate() {
 
         if (checkpointConfig.isCheckpointingEnabled()) {
             // temporarily forbid checkpointing for iterative jobs
-            if (streamGraph.isIterative() && !checkpointConfig.isForceCheckpointing()) {
-                throw new UnsupportedOperationException(
-                        "Checkpointing is currently not supported by default for iterative jobs, as we cannot guarantee exactly once semantics. "
-                                + "State checkpoints happen normally, but records in-transit during the snapshot will be lost upon failure. "
-                                + "\nThe user can force enable state checkpoints with the reduced guarantees by calling: env.enableCheckpointing(interval,true)");
-            }
             if (streamGraph.isIterative()
                     && checkpointConfig.isUnalignedCheckpointsEnabled()
                     && !checkpointConfig.isForceUnalignedCheckpoints()) {

flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamExecutionEnvironment.scala

@@ -177,34 +177,6 @@ class StreamExecutionEnvironment(javaEnv: JavaEnv) extends AutoCloseable {
    */
   def getCheckpointConfig = javaEnv.getCheckpointConfig()
 
-  /**
-   * Enables checkpointing for the streaming job. The distributed state of the streaming dataflow
-   * will be periodically snapshotted. In case of a failure, the streaming dataflow will be
-   * restarted from the latest completed checkpoint.
-   *
-   * The job draws checkpoints periodically, in the given interval. The state will be stored in the
-   * configured state backend.
-   *
-   * NOTE: Checkpointing iterative streaming dataflows in not properly supported at the moment. If
-   * the "force" parameter is set to true, the system will execute the job nonetheless.
-   *
-   * @param interval
-   *   Time interval between state checkpoints in millis.
-   * @param mode
-   *   The checkpointing mode, selecting between "exactly once" and "at least once" guarantees.
-   * @param force
-   *   If true checkpointing will be enabled for iterative jobs as well.
-   */
-  @deprecated
-  @PublicEvolving
-  def enableCheckpointing(
-      interval: Long,
-      mode: org.apache.flink.streaming.api.CheckpointingMode,
-      force: Boolean): StreamExecutionEnvironment = {
-    javaEnv.enableCheckpointing(interval, mode, force)
-    this
-  }
-
   /**
    * Enables checkpointing for the streaming job. The distributed state of the streaming dataflow
    * will be periodically snapshotted. In case of a failure, the streaming dataflow will be

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/utils/DummyStreamExecutionEnvironment.java

@@ -162,13 +162,6 @@ public StreamExecutionEnvironment enableCheckpointing(long interval, Checkpointi
                 "This is a dummy StreamExecutionEnvironment, enableCheckpointing method is unsupported.");
     }
 
-    @Override
-    public StreamExecutionEnvironment enableCheckpointing(
-            long interval, org.apache.flink.streaming.api.CheckpointingMode mode, boolean force) {
-        throw new UnsupportedOperationException(
-                "This is a dummy StreamExecutionEnvironment, enableCheckpointing method is unsupported.");
-    }
-
     @Override
     public StreamExecutionEnvironment enableCheckpointing() {
         throw new UnsupportedOperationException(
@@ -180,11 +173,6 @@ public long getCheckpointInterval() {
         return realExecEnv.getCheckpointInterval();
     }
 
-    @Override
-    public boolean isForceCheckpointing() {
-        return realExecEnv.isForceCheckpointing();
-    }
-
     @Override
     public org.apache.flink.streaming.api.CheckpointingMode getCheckpointingMode() {
         return realExecEnv.getCheckpointingMode();

flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/IterateITCase.java

@@ -25,7 +25,6 @@
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.runtime.jobgraph.JobGraph;
 import org.apache.flink.runtime.jobgraph.JobVertex;
-import org.apache.flink.runtime.jobgraph.tasks.CheckpointCoordinatorConfiguration;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.datastream.DataStreamSink;
 import org.apache.flink.streaming.api.datastream.IterativeStream;
@@ -631,63 +630,6 @@ public void close() {
         }
     }
 
-    @SuppressWarnings("deprecation")
-    @Test
-    public void testWithCheckPointing() throws Exception {
-        int numRetries = 5;
-        int timeoutScale = 1;
-
-        for (int numRetry = 0; numRetry < numRetries; numRetry++) {
-            try {
-                StreamExecutionEnvironment env =
-                        StreamExecutionEnvironment.getExecutionEnvironment();
-
-                try {
-                    createIteration(env, timeoutScale);
-                    env.execute();
-
-                    // this statement should never be reached
-                    fail();
-                } catch (UnsupportedOperationException e) {
-                    // expected behaviour
-                }
-
-                // Test force checkpointing
-
-                try {
-                    createIteration(env, timeoutScale);
-                    env.enableCheckpointing(
-                            CheckpointCoordinatorConfiguration.MINIMAL_CHECKPOINT_TIME,
-                            org.apache.flink.streaming.api.CheckpointingMode.EXACTLY_ONCE,
-                            false);
-                    env.execute();
-
-                    // this statement should never be reached
-                    fail();
-                } catch (UnsupportedOperationException e) {
-                    // expected behaviour
-                }
-
-                createIteration(env, timeoutScale);
-                env.enableCheckpointing(
-                        CheckpointCoordinatorConfiguration.MINIMAL_CHECKPOINT_TIME,
-                        org.apache.flink.streaming.api.CheckpointingMode.EXACTLY_ONCE,
-                        true);
-                env.getStreamGraph().getJobGraph();
-
-                break; // success
-            } catch (Throwable t) {
-                LOG.info("Run " + (numRetry + 1) + "/" + numRetries + " failed", t);
-
-                if (numRetry >= numRetries - 1) {
-                    throw t;
-                } else {
-                    timeoutScale *= 2;
-                }
-            }
-        }
-    }
-
     private void createIteration(StreamExecutionEnvironment env, int timeoutScale) {
         env.enableCheckpointing();