You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by dw...@apache.org on 2021/06/08 06:31:39 UTC
[flink] 01/03: [FLINK-22815][checkpointing] Disable unaligned
checkpoints for broadcast partitioning
This is an automated email from the ASF dual-hosted git repository.
dwysakowicz pushed a commit to branch release-1.12
in repository https://gitbox.apache.org/repos/asf/flink.git
commit bdf43992f6761bd994055b3c4fb2f462257cbf5f
Author: Dawid Wysakowicz <dw...@apache.org>
AuthorDate: Fri May 28 09:39:09 2021 +0200
[FLINK-22815][checkpointing] Disable unaligned checkpoints for broadcast partitioning
Broadcast partitioning can not work with unaligned checkpointing. There
is no guarantees that records are consumed at the same rate in all
channels. This can result in some tasks applying state changes
corresponding to a certain broadcasted event while others don't. In turn
upon restore it may lead to an unconsistent state.
---
.../io/network/api/writer/SubtaskStateMapper.java | 11 ++++
.../flink/streaming/api/graph/StreamEdge.java | 10 ++++
.../streaming/api/graph/StreamGraphGenerator.java | 8 +++
.../api/graph/StreamGraphGeneratorTest.java | 65 +++++++++++++++++++++-
4 files changed, 93 insertions(+), 1 deletion(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/SubtaskStateMapper.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/SubtaskStateMapper.java
index b7de773..7bb5611 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/SubtaskStateMapper.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/SubtaskStateMapper.java
@@ -180,6 +180,17 @@ public enum SubtaskStateMapper {
}
return subtasks;
}
+ },
+
+ UNSUPPORTED {
+ @Override
+ public Set<Integer> getOldSubtasks(
+ int newSubtaskIndex, int oldNumberOfSubtasks, int newNumberOfSubtasks) {
+ throw new UnsupportedOperationException(
+ "Cannot rescale the given pointwise partitioner.\n"
+ + "Did you change the partitioner to forward or rescale?\n"
+ + "It may also help to add an explicit shuffle().");
+ }
};
/**
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamEdge.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamEdge.java
index f8a2027..f76edd5 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamEdge.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamEdge.java
@@ -62,6 +62,8 @@ public class StreamEdge implements Serializable {
private long bufferTimeout;
+ private boolean supportsUnalignedCheckpoints = true;
+
public StreamEdge(
StreamNode sourceVertex,
StreamNode targetVertex,
@@ -156,6 +158,14 @@ public class StreamEdge implements Serializable {
return bufferTimeout;
}
+ public void setSupportsUnalignedCheckpoints(boolean supportsUnalignedCheckpoints) {
+ this.supportsUnalignedCheckpoints = supportsUnalignedCheckpoints;
+ }
+
+ public boolean supportsUnalignedCheckpoints() {
+ return supportsUnalignedCheckpoints;
+ }
+
@Override
public int hashCode() {
return Objects.hash(edgeId, outputTag);
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraphGenerator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraphGenerator.java
index a9917d8..27c0eaf 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraphGenerator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraphGenerator.java
@@ -270,6 +270,14 @@ public class StreamGraphGenerator {
transform(transformation);
}
+ for (StreamNode node : streamGraph.getStreamNodes()) {
+ if (node.getInEdges().stream().anyMatch(edge -> edge.getPartitioner().isBroadcast())) {
+ for (StreamEdge edge : node.getInEdges()) {
+ edge.setSupportsUnalignedCheckpoints(false);
+ }
+ }
+ }
+
final StreamGraph builtStreamGraph = streamGraph;
alreadyTransformed.clear();
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamGraphGeneratorTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamGraphGeneratorTest.java
index 9e8fb13..24b6363 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamGraphGeneratorTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamGraphGeneratorTest.java
@@ -20,6 +20,7 @@ package org.apache.flink.streaming.api.graph;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.operators.ResourceSpec;
+import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.dag.Transformation;
@@ -28,6 +29,7 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.memory.ManagedMemoryUseCase;
import org.apache.flink.runtime.jobgraph.SavepointConfigOptions;
import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
+import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.IterativeStream;
@@ -35,6 +37,7 @@ import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;
+import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
@@ -58,9 +61,11 @@ import org.apache.flink.streaming.runtime.streamrecord.LatencyMarker;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.runtime.tasks.StreamTask;
import org.apache.flink.streaming.util.NoOpIntMap;
+import org.apache.flink.util.Collector;
import org.apache.flink.util.TestLogger;
import org.hamcrest.Description;
+import org.hamcrest.FeatureMatcher;
import org.hamcrest.Matcher;
import org.hamcrest.TypeSafeMatcher;
import org.junit.Test;
@@ -70,11 +75,12 @@ import java.util.Collection;
import java.util.Collections;
import java.util.List;
+import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.iterableWithSize;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
/**
@@ -289,6 +295,63 @@ public class StreamGraphGeneratorTest extends TestLogger {
assertEquals(0, streamGraph.getStreamEdges(transform.getId()).size());
}
+ @Test
+ public void testUnalignedCheckpointDisabledOnBroadcast() {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(42);
+
+ DataStream<Long> source1 = env.fromSequence(1L, 10L);
+ DataStream<Long> map1 = source1.broadcast().map(l -> l);
+ DataStream<Long> source2 = env.fromSequence(2L, 11L);
+ DataStream<Long> keyed = source2.keyBy(r -> 0L);
+
+ final MapStateDescriptor<Long, Long> descriptor =
+ new MapStateDescriptor<>(
+ "broadcast", BasicTypeInfo.LONG_TYPE_INFO, BasicTypeInfo.LONG_TYPE_INFO);
+ final BroadcastStream<Long> broadcast = map1.broadcast(descriptor);
+ final SingleOutputStreamOperator<Long> joined =
+ keyed.connect(broadcast)
+ .process(
+ new KeyedBroadcastProcessFunction<Long, Long, Long, Long>() {
+ @Override
+ public void processElement(
+ Long value, ReadOnlyContext ctx, Collector<Long> out) {}
+
+ @Override
+ public void processBroadcastElement(
+ Long value, Context ctx, Collector<Long> out) {}
+ });
+
+ StreamGraph streamGraph = env.getStreamGraph();
+ assertEquals(4, streamGraph.getStreamNodes().size());
+
+ // single broadcast
+ assertThat(edge(streamGraph, source1, map1), supportsUnalignedCheckpoints(false));
+ // keyed, connected with broadcast
+ assertThat(edge(streamGraph, source2, joined), supportsUnalignedCheckpoints(false));
+ // broadcast, connected with keyed
+ assertThat(edge(streamGraph, map1, joined), supportsUnalignedCheckpoints(false));
+ }
+
+ private static StreamEdge edge(
+ StreamGraph streamGraph, DataStream<Long> op1, DataStream<Long> op2) {
+ List<StreamEdge> streamEdges = streamGraph.getStreamEdges(op1.getId(), op2.getId());
+ assertThat(streamEdges, iterableWithSize(1));
+ return streamEdges.get(0);
+ }
+
+ private static Matcher<StreamEdge> supportsUnalignedCheckpoints(boolean enabled) {
+ return new FeatureMatcher<StreamEdge, Boolean>(
+ equalTo(enabled),
+ "supports unaligned checkpoint",
+ "supports unaligned checkpoint") {
+ @Override
+ protected Boolean featureValueOf(StreamEdge actual) {
+ return actual.supportsUnalignedCheckpoints();
+ }
+ };
+ }
+
/**
* Tests that the KeyGroupStreamPartitioner are properly set up with the correct value of
* maximum parallelism.