You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by zh...@apache.org on 2020/04/13 15:18:15 UTC
[flink] 01/02: [FLINK-17107][checkpointing] Make
CheckpointCoordinatorConfiguration#isExactlyOnce() be consistent with
StreamConfig#getCheckpointMode()
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit eb4a2115db91cf2908312e8c4ddc673f6c84e6bd
Author: kevin.cyj <ke...@alibaba-inc.com>
AuthorDate: Mon Apr 13 16:45:42 2020 +0800
[FLINK-17107][checkpointing] Make CheckpointCoordinatorConfiguration#isExactlyOnce() be consistent with StreamConfig#getCheckpointMode()
---
.../streaming/api/graph/StreamingJobGraphGenerator.java | 2 +-
.../streaming/api/graph/StreamingJobGraphGeneratorTest.java | 12 ++++++++++--
2 files changed, 11 insertions(+), 3 deletions(-)
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
index 5a34863..410ee3d 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
@@ -885,7 +885,7 @@ public class StreamingJobGraphGenerator {
boolean isExactlyOnce;
if (mode == CheckpointingMode.EXACTLY_ONCE) {
- isExactlyOnce = true;
+ isExactlyOnce = cfg.isCheckpointingEnabled();
} else if (mode == CheckpointingMode.AT_LEAST_ONCE) {
isExactlyOnce = false;
} else {
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGeneratorTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGeneratorTest.java
index 7ce05c4..4eb15a6 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGeneratorTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGeneratorTest.java
@@ -44,6 +44,7 @@ import org.apache.flink.runtime.jobgraph.tasks.JobCheckpointingSettings;
import org.apache.flink.runtime.jobmanager.scheduler.CoLocationGroup;
import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup;
import org.apache.flink.runtime.operators.util.TaskConfig;
+import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
@@ -156,18 +157,25 @@ public class StreamingJobGraphGeneratorTest extends TestLogger {
}
/**
- * Tests that disabled checkpointing sets the checkpointing interval to Long.MAX_VALUE.
+ * Tests that disabled checkpointing sets the checkpointing interval to Long.MAX_VALUE and the checkpoint mode to
+ * {@link CheckpointingMode#AT_LEAST_ONCE}.
*/
@Test
public void testDisabledCheckpointing() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
- StreamGraph streamGraph = new StreamGraph(env.getConfig(), env.getCheckpointConfig(), SavepointRestoreSettings.none());
+ env.fromElements(0).print();
+ StreamGraph streamGraph = env.getStreamGraph();
assertFalse("Checkpointing enabled", streamGraph.getCheckpointConfig().isCheckpointingEnabled());
JobGraph jobGraph = StreamingJobGraphGenerator.createJobGraph(streamGraph);
JobCheckpointingSettings snapshottingSettings = jobGraph.getCheckpointingSettings();
assertEquals(Long.MAX_VALUE, snapshottingSettings.getCheckpointCoordinatorConfiguration().getCheckpointInterval());
+ assertFalse(snapshottingSettings.getCheckpointCoordinatorConfiguration().isExactlyOnce());
+
+ List<JobVertex> verticesSorted = jobGraph.getVerticesSortedTopologicallyFromSources();
+ StreamConfig streamConfig = new StreamConfig(verticesSorted.get(0).getConfiguration());
+ assertEquals(CheckpointingMode.AT_LEAST_ONCE, streamConfig.getCheckpointMode());
}
@Test