[FLINK-25528][state-processor-api] state processor api do not support increment checkpoint

1. limit the checkpoint type in the savepoint type and full checkpoint type.
2. add the SnapshotUtils test with the full checkpoint.

Author: liufangqi

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

@@ -23,6 +23,7 @@
 import org.apache.flink.core.execution.SavepointFormatType;
 import org.apache.flink.core.fs.Path;
 import org.apache.flink.runtime.checkpoint.CheckpointOptions;
+import org.apache.flink.runtime.checkpoint.CheckpointType;
 import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
 import org.apache.flink.runtime.checkpoint.SavepointType;
 import org.apache.flink.runtime.checkpoint.SnapshotType;
@@ -33,6 +34,7 @@
 import org.apache.flink.streaming.api.operators.OperatorSnapshotFutures;
 import org.apache.flink.streaming.api.operators.StreamOperator;
 import org.apache.flink.util.MathUtils;
+import org.apache.flink.util.Preconditions;
 
 import java.io.IOException;
 
@@ -57,6 +59,10 @@ public static <OUT, OP extends StreamOperator<OUT>> TaggedOperatorSubtaskState s
             SnapshotType snapshotType)
             throws Exception {
 
+        Preconditions.checkArgument(
+                snapshotType.isSavepoint() || CheckpointType.FULL_CHECKPOINT.equals(snapshotType),
+                "the snapshot type require savepoint type or full checkpoint type.");
+
         CheckpointOptions options =
                 CheckpointOptions.forConfig(
                         snapshotType,

flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/output/SnapshotUtilsTest.java

@@ -22,6 +22,7 @@
 import org.apache.flink.core.fs.Path;
 import org.apache.flink.metrics.groups.OperatorMetricGroup;
 import org.apache.flink.runtime.checkpoint.CheckpointOptions;
+import org.apache.flink.runtime.checkpoint.CheckpointType;
 import org.apache.flink.runtime.jobgraph.OperatorID;
 import org.apache.flink.runtime.state.CheckpointStreamFactory;
 import org.apache.flink.streaming.api.operators.OperatorSnapshotFutures;
@@ -52,6 +53,7 @@ public class SnapshotUtilsTest {
 
     @Test
     public void testSnapshotUtilsLifecycle() throws Exception {
+        ACTUAL_ORDER_TRACKING.clear();
         StreamOperator<Void> operator = new LifecycleOperator();
         Path path = new Path(folder.newFolder().getAbsolutePath());
 
@@ -60,6 +62,25 @@ public void testSnapshotUtilsLifecycle() throws Exception {
         Assert.assertEquals(EXPECTED_CALL_OPERATOR_SNAPSHOT, ACTUAL_ORDER_TRACKING);
     }
 
+    @Test
+    public void testSnapshotUtilsLifecycleWithFullCheckpoint() throws Exception {
+        ACTUAL_ORDER_TRACKING.clear();
+        StreamOperator<Void> operator = new LifecycleOperator();
+        Path path = new Path(folder.newFolder().getAbsolutePath());
+
+        SnapshotUtils.snapshot(
+                operator,
+                0,
+                0L,
+                true,
+                false,
+                new Configuration(),
+                path,
+                CheckpointType.FULL_CHECKPOINT);
+
+        Assert.assertEquals(EXPECTED_CALL_OPERATOR_SNAPSHOT, ACTUAL_ORDER_TRACKING);
+    }
+
     private static class LifecycleOperator implements StreamOperator<Void> {
         private static final long serialVersionUID = 1L;