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 2023/02/07 11:17:38 UTC
[flink] branch release-1.17 updated: [FLINK-30901][runtime] Fix the jobVertex's parallelismConfigured is incorrect when chaining with source operators
This is an automated email from the ASF dual-hosted git repository.
zhuzh pushed a commit to branch release-1.17
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.17 by this push:
new 5e198e57df3 [FLINK-30901][runtime] Fix the jobVertex's parallelismConfigured is incorrect when chaining with source operators
5e198e57df3 is described below
commit 5e198e57df35a9bc0b979f9e79e22872e753211c
Author: JunRuiLee <jr...@gmail.com>
AuthorDate: Tue Feb 7 11:14:37 2023 +0800
[FLINK-30901][runtime] Fix the jobVertex's parallelismConfigured is incorrect when chaining with source operators
This closes #21854.
---
.../api/graph/StreamingJobGraphGenerator.java | 24 +++++------
.../api/graph/StreamingJobGraphGeneratorTest.java | 48 +++++++++++++++++++++-
2 files changed, 58 insertions(+), 14 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 cd78954990f..cbb02e5a164 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
@@ -596,6 +596,7 @@ public class StreamingJobGraphGenerator {
chainedSources,
streamGraph));
chainInfo.addCoordinatorProvider(coord);
+ chainInfo.recordChainedNode(sourceNodeId);
continue;
}
}
@@ -898,7 +899,10 @@ public class StreamingJobGraphGenerator {
jobVertices.put(streamNodeId, jobVertex);
builtVertices.add(streamNodeId);
jobGraph.addVertex(jobVertex);
- jobVertex.setParallelismConfigured(chainInfo.isParallelismConfigured());
+
+ jobVertex.setParallelismConfigured(
+ chainInfo.getAllChainedNodes().stream()
+ .anyMatch(StreamNode::isParallelismConfigured));
if (streamGraph.isDynamic()
&& !jobVertex.isParallelismConfigured()
&& streamGraph.isAutoParallelismEnabled()) {
@@ -1835,7 +1839,6 @@ public class StreamingJobGraphGenerator {
private final Map<Integer, ChainedSourceInfo> chainedSources;
private final List<OperatorCoordinator.Provider> coordinatorProviders;
private final StreamGraph streamGraph;
- private boolean parallelismConfigured;
private final List<StreamNode> chainedNodes;
private OperatorChainInfo(
@@ -1851,10 +1854,7 @@ public class StreamingJobGraphGenerator {
this.coordinatorProviders = new ArrayList<>();
this.chainedSources = chainedSources;
this.streamGraph = streamGraph;
- this.parallelismConfigured =
- streamGraph.getStreamNode(startNodeId).isParallelismConfigured();
this.chainedNodes = new ArrayList<>();
- chainedNodes.add(streamGraph.getStreamNode(startNodeId));
}
byte[] getHash(Integer streamNodeId) {
@@ -1881,14 +1881,9 @@ public class StreamingJobGraphGenerator {
return chainedSources;
}
- boolean isParallelismConfigured() {
- return parallelismConfigured;
- }
-
private OperatorID addNodeToChain(int currentNodeId, String operatorName) {
+ recordChainedNode(currentNodeId);
StreamNode streamNode = streamGraph.getStreamNode(currentNodeId);
- chainedNodes.add(streamNode);
- parallelismConfigured |= streamNode.isParallelismConfigured();
List<Tuple2<byte[], byte[]>> operatorHashes =
chainedOperatorHashes.computeIfAbsent(startNodeId, k -> new ArrayList<>());
@@ -1906,12 +1901,17 @@ public class StreamingJobGraphGenerator {
return new OperatorID(primaryHashBytes);
}
+ private void recordChainedNode(int currentNodeId) {
+ StreamNode streamNode = streamGraph.getStreamNode(currentNodeId);
+ chainedNodes.add(streamNode);
+ }
+
private OperatorChainInfo newChain(Integer startNodeId) {
return new OperatorChainInfo(
startNodeId, hashes, legacyHashes, chainedSources, streamGraph);
}
- public List<StreamNode> getAllChainedNodes() {
+ private List<StreamNode> getAllChainedNodes() {
return chainedNodes;
}
}
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 67969d8c512..6018c380ea3 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
@@ -259,7 +259,7 @@ class StreamingJobGraphGeneratorTest {
env.fromSequence(1L, 3L).map(i -> i).setParallelism(10).print().setParallelism(20);
StreamGraph streamGraph = env.getStreamGraph();
- // check the streamGraph parallleism configured
+ // check the streamGraph parallelism configured
final List<StreamNode> streamNodes =
streamGraph.getStreamNodes().stream()
.sorted(Comparator.comparingInt(StreamNode::getId))
@@ -283,7 +283,7 @@ class StreamingJobGraphGeneratorTest {
env.fromSequence(1L, 3L).map(value -> value).print().setParallelism(env.getParallelism());
StreamGraph streamGraph = env.getStreamGraph();
- // check the streamGraph parallleism configured
+ // check the streamGraph parallelism configured
final List<StreamNode> streamNodes =
streamGraph.getStreamNodes().stream()
.sorted(Comparator.comparingInt(StreamNode::getId))
@@ -299,6 +299,50 @@ class StreamingJobGraphGeneratorTest {
assertThat(vertices.get(0).isParallelismConfigured()).isTrue();
}
+ @Test
+ public void testChainedSourcesSetParallelism() {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ MultipleInputTransformation<Long> transform =
+ new MultipleInputTransformation<>(
+ "mit",
+ new UnusedOperatorFactory(),
+ Types.LONG,
+ env.getParallelism(),
+ false);
+ DataStreamSource<Long> source1 =
+ env.fromSource(
+ new NumberSequenceSource(1, 2),
+ WatermarkStrategy.noWatermarks(),
+ "source1");
+ DataStreamSource<Long> source2 =
+ env.fromSource(
+ new NumberSequenceSource(1, 2),
+ WatermarkStrategy.noWatermarks(),
+ "source2");
+ transform.addInput(source1.getTransformation());
+ transform.addInput(source2.getTransformation());
+ transform.setChainingStrategy(ChainingStrategy.HEAD_WITH_SOURCES);
+ source1.setParallelism(env.getParallelism());
+ env.addOperator(transform);
+
+ StreamGraph streamGraph = env.getStreamGraph();
+
+ // check the streamGraph parallelism configured
+ final List<StreamNode> streamNodes =
+ streamGraph.getStreamNodes().stream()
+ .sorted(Comparator.comparingInt(StreamNode::getId))
+ .collect(Collectors.toList());
+ assertThat(streamNodes.get(0).isParallelismConfigured()).isFalse();
+ assertThat(streamNodes.get(1).isParallelismConfigured()).isTrue();
+ assertThat(streamNodes.get(2).isParallelismConfigured()).isFalse();
+
+ // check the jobGraph parallelism configured
+ JobGraph jobGraph = StreamingJobGraphGenerator.createJobGraph(streamGraph);
+ List<JobVertex> vertices = jobGraph.getVerticesSortedTopologicallyFromSources();
+ assertThat(jobGraph.getNumberOfVertices()).isEqualTo(1);
+ assertThat(vertices.get(0).isParallelismConfigured()).isTrue();
+ }
+
@Test
public void testDynamicGraphVertexParallelism() {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();