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:14 UTC
[flink] branch master updated (39ebaae -> 531d6bf)
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git.
from 39ebaae [FLINK-17050][runtime] Rename methods getVertexOrThrow() and getResultPartitionOrThrow()
new eb4a211 [FLINK-17107][checkpointing] Make CheckpointCoordinatorConfiguration#isExactlyOnce() be consistent with StreamConfig#getCheckpointMode()
new 531d6bf [hotfix] Extract common logic of getting CheckpointingMode as StreamingJobGraphGenerator#getCheckpointingMode
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../api/graph/StreamingJobGraphGenerator.java | 41 ++++++++++------------
.../api/graph/StreamingJobGraphGeneratorTest.java | 12 +++++--
2 files changed, 29 insertions(+), 24 deletions(-)
[flink] 02/02: [hotfix] Extract common logic of getting
CheckpointingMode as StreamingJobGraphGenerator#getCheckpointingMode
Posted by zh...@apache.org.
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 531d6bf4af301616c5d14786535987be24c0e27f
Author: kevin.cyj <ke...@alibaba-inc.com>
AuthorDate: Mon Apr 13 16:57:41 2020 +0800
[hotfix] Extract common logic of getting CheckpointingMode as StreamingJobGraphGenerator#getCheckpointingMode
---
.../api/graph/StreamingJobGraphGenerator.java | 41 ++++++++++------------
1 file changed, 19 insertions(+), 22 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 410ee3d..6e22b73 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
@@ -84,6 +84,7 @@ import java.util.Map.Entry;
import java.util.Set;
import static org.apache.flink.runtime.jobgraph.tasks.CheckpointCoordinatorConfiguration.MINIMAL_CHECKPOINT_TIME;
+import static org.apache.flink.util.Preconditions.checkArgument;
import static org.apache.flink.util.Preconditions.checkState;
/**
@@ -516,14 +517,8 @@ public class StreamingJobGraphGenerator {
config.setStateBackend(streamGraph.getStateBackend());
config.setCheckpointingEnabled(checkpointCfg.isCheckpointingEnabled());
- if (checkpointCfg.isCheckpointingEnabled()) {
- config.setCheckpointMode(checkpointCfg.getCheckpointingMode());
- }
- else {
- // the "at-least-once" input handler is slightly cheaper (in the absence of checkpoints),
- // so we use that one if checkpointing is not enabled
- config.setCheckpointMode(CheckpointingMode.AT_LEAST_ONCE);
- }
+ config.setCheckpointMode(getCheckpointingMode(checkpointCfg));
+
for (int i = 0; i < vertex.getStatePartitioners().length; i++) {
config.setStatePartitioner(i, vertex.getStatePartitioners()[i]);
}
@@ -540,6 +535,21 @@ public class StreamingJobGraphGenerator {
vertexConfigs.put(vertexID, config);
}
+ private CheckpointingMode getCheckpointingMode(CheckpointConfig checkpointConfig) {
+ CheckpointingMode checkpointingMode = checkpointConfig.getCheckpointingMode();
+
+ checkArgument(checkpointingMode == CheckpointingMode.EXACTLY_ONCE ||
+ checkpointingMode == CheckpointingMode.AT_LEAST_ONCE, "Unexpected checkpointing mode.");
+
+ if (checkpointConfig.isCheckpointingEnabled()) {
+ return checkpointingMode;
+ } else {
+ // the "at-least-once" input handler is slightly cheaper (in the absence of checkpoints),
+ // so we use that one if checkpointing is not enabled
+ return CheckpointingMode.AT_LEAST_ONCE;
+ }
+ }
+
private void connect(Integer headOfChain, StreamEdge edge) {
physicalEdgesInOrder.add(edge);
@@ -881,19 +891,6 @@ public class StreamingJobGraphGenerator {
retentionAfterTermination = CheckpointRetentionPolicy.NEVER_RETAIN_AFTER_TERMINATION;
}
- CheckpointingMode mode = cfg.getCheckpointingMode();
-
- boolean isExactlyOnce;
- if (mode == CheckpointingMode.EXACTLY_ONCE) {
- isExactlyOnce = cfg.isCheckpointingEnabled();
- } else if (mode == CheckpointingMode.AT_LEAST_ONCE) {
- isExactlyOnce = false;
- } else {
- throw new IllegalStateException("Unexpected checkpointing mode. " +
- "Did not expect there to be another checkpointing mode besides " +
- "exactly-once or at-least-once.");
- }
-
// --- configure the master-side checkpoint hooks ---
final ArrayList<MasterTriggerRestoreHook.Factory> hooks = new ArrayList<>();
@@ -951,7 +948,7 @@ public class StreamingJobGraphGenerator {
cfg.getMinPauseBetweenCheckpoints(),
cfg.getMaxConcurrentCheckpoints(),
retentionAfterTermination,
- isExactlyOnce,
+ getCheckpointingMode(cfg) == CheckpointingMode.EXACTLY_ONCE,
cfg.isPreferCheckpointForRecovery(),
cfg.getTolerableCheckpointFailureNumber()),
serializedStateBackend,
[flink] 01/02: [FLINK-17107][checkpointing] Make
CheckpointCoordinatorConfiguration#isExactlyOnce() be consistent with
StreamConfig#getCheckpointMode()
Posted by zh...@apache.org.
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