[FLINK-8531] [checkpoints] (part 1) Pull CheckpointType into its own …

…class.

flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java

@@ -45,6 +45,7 @@
 import org.apache.flink.core.memory.DataOutputView;
 import org.apache.flink.core.memory.DataOutputViewStreamWrapper;
 import org.apache.flink.runtime.checkpoint.CheckpointOptions;
+import org.apache.flink.runtime.checkpoint.CheckpointType;
 import org.apache.flink.runtime.io.async.AbstractAsyncCallableWithResources;
 import org.apache.flink.runtime.io.async.AsyncStoppableTaskWithCallback;
 import org.apache.flink.runtime.query.TaskKvStateRegistry;
@@ -343,7 +344,7 @@ public RunnableFuture<KeyedStateHandle> snapshot(
  final CheckpointStreamFactory streamFactory,
  CheckpointOptions checkpointOptions) throws Exception {
 
- if (checkpointOptions.getCheckpointType() != CheckpointOptions.CheckpointType.SAVEPOINT &&
+ if (checkpointOptions.getCheckpointType() != CheckpointType.SAVEPOINT &&
  enableIncrementalCheckpointing) {
  return snapshotIncrementally(checkpointId, timestamp, streamFactory);
  } else {

flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointOptions.java

@@ -93,17 +93,4 @@ public static CheckpointOptions forSavepoint(String targetDirectory) {
 
  // ------------------------------------------------------------------------
 
- /**
- * The type of checkpoint to perform.
- */
- public enum CheckpointType {
-
- /** A full checkpoint. */
- CHECKPOINT,
-
- /** A savepoint. */
- SAVEPOINT;
-
- }
-
 }

flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointType.java

@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.checkpoint;
+
+/**
+ * The type of checkpoint to perform.
+ */
+public enum CheckpointType {
+
+ /** A checkpoint, full or incremental. */
+ CHECKPOINT,
+
+ /** A savepoint. */
+ SAVEPOINT;
+
+}

flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/EventSerializer.java

@@ -22,7 +22,7 @@
 import org.apache.flink.core.memory.MemorySegment;
 import org.apache.flink.core.memory.MemorySegmentFactory;
 import org.apache.flink.runtime.checkpoint.CheckpointOptions;
-import org.apache.flink.runtime.checkpoint.CheckpointOptions.CheckpointType;
+import org.apache.flink.runtime.checkpoint.CheckpointType;
 import org.apache.flink.runtime.event.AbstractEvent;
 import org.apache.flink.runtime.io.network.api.CancelCheckpointMarker;
 import org.apache.flink.runtime.io.network.api.CheckpointBarrier;

flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointOptionsTest.java

@@ -21,7 +21,6 @@
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNull;
 
-import org.apache.flink.runtime.checkpoint.CheckpointOptions.CheckpointType;
 import org.junit.Test;
 
 public class CheckpointOptionsTest {

flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointTypeTest.java

@@ -18,8 +18,6 @@
 
 package org.apache.flink.runtime.checkpoint;
 
-import org.apache.flink.runtime.checkpoint.CheckpointOptions.CheckpointType;
-
 import org.junit.Test;
 
 import static org.junit.Assert.assertEquals;

flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/JobManagerTest.java

@@ -33,7 +33,7 @@
 import org.apache.flink.runtime.checkpoint.CheckpointMetaData;
 import org.apache.flink.runtime.checkpoint.CheckpointMetrics;
 import org.apache.flink.runtime.checkpoint.CheckpointOptions;
-import org.apache.flink.runtime.checkpoint.CheckpointOptions.CheckpointType;
+import org.apache.flink.runtime.checkpoint.CheckpointType;
 import org.apache.flink.runtime.checkpoint.CheckpointRetentionPolicy;
 import org.apache.flink.runtime.clusterframework.messages.NotifyResourceStarted;
 import org.apache.flink.runtime.clusterframework.messages.RegisterResourceManager;

flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManagerLike.scala

@@ -26,9 +26,8 @@ import akka.pattern.{ask, pipe}
 import org.apache.flink.api.common.JobID
 import org.apache.flink.core.fs.FSDataInputStream
 import org.apache.flink.runtime.FlinkActor
-import org.apache.flink.runtime.checkpoint.CheckpointOptions.CheckpointType
 import org.apache.flink.runtime.checkpoint.savepoint.Savepoint
-import org.apache.flink.runtime.checkpoint.{Checkpoints, CompletedCheckpoint}
+import org.apache.flink.runtime.checkpoint.{CheckpointType, Checkpoints, CompletedCheckpoint}
 import org.apache.flink.runtime.execution.ExecutionState
 import org.apache.flink.runtime.jobgraph.JobStatus
 import org.apache.flink.runtime.jobmanager.JobManager

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java

@@ -34,7 +34,7 @@
 import org.apache.flink.metrics.Gauge;
 import org.apache.flink.metrics.MetricGroup;
 import org.apache.flink.runtime.checkpoint.CheckpointOptions;
-import org.apache.flink.runtime.checkpoint.CheckpointOptions.CheckpointType;
+import org.apache.flink.runtime.checkpoint.CheckpointType;
 import org.apache.flink.runtime.execution.Environment;
 import org.apache.flink.runtime.jobgraph.OperatorID;
 import org.apache.flink.runtime.metrics.groups.OperatorMetricGroup;