You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by xt...@apache.org on 2023/02/02 09:11:56 UTC
[flink] 01/05: [hotfix] Validate hybrid shuffle mode only execute in batch mode.
This is an automated email from the ASF dual-hosted git repository.
xtsong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 80b755f803d9609eeac83327cf0c934d51a7233f
Author: Weijie Guo <re...@163.com>
AuthorDate: Fri Jan 13 16:44:12 2023 +0800
[hotfix] Validate hybrid shuffle mode only execute in batch mode.
---
.../api/graph/StreamingJobGraphGenerator.java | 14 ++++++++++++
.../api/graph/StreamingJobGraphGeneratorTest.java | 26 ++++++++++++++++++++++
2 files changed, 40 insertions(+)
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 c960664e840..cd78954990f 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
@@ -21,6 +21,7 @@ import org.apache.flink.annotation.Internal;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.cache.DistributedCache;
import org.apache.flink.api.common.functions.Function;
import org.apache.flink.api.common.operators.ResourceSpec;
@@ -41,6 +42,7 @@ import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
import org.apache.flink.runtime.jobgraph.JobEdge;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobgraph.JobGraphUtils;
+import org.apache.flink.runtime.jobgraph.JobType;
import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobgraph.OperatorID;
@@ -243,6 +245,8 @@ public class StreamingJobGraphGenerator {
markSupportingConcurrentExecutionAttempts();
+ validateHybridShuffleExecuteInBatchMode();
+
setSlotSharingAndCoLocation();
setManagedMemoryFraction(
@@ -1467,6 +1471,16 @@ public class StreamingJobGraphGenerator {
}
}
+ private void validateHybridShuffleExecuteInBatchMode() {
+ if (hasHybridResultPartition) {
+ checkState(
+ jobGraph.getJobType() == JobType.BATCH,
+ "hybrid shuffle mode only supports batch job, please set %s to %s",
+ ExecutionOptions.RUNTIME_MODE.key(),
+ RuntimeExecutionMode.BATCH.name());
+ }
+ }
+
/**
* Maps a vertex to its region slot sharing group. If {@link
* StreamGraph#isAllVerticesInSameSlotSharingGroupByDefault()} returns true, all regions will be
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 7e81cbda762..67969d8c512 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
@@ -813,6 +813,7 @@ class StreamingJobGraphGeneratorTest {
@Test
void testExchangeModeHybridFull() {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setRuntimeMode(RuntimeExecutionMode.BATCH);
// fromElements -> Map -> Print
DataStream<Integer> sourceDataStream = env.fromElements(1, 2, 3);
@@ -852,6 +853,7 @@ class StreamingJobGraphGeneratorTest {
@Test
void testExchangeModeHybridSelective() {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setRuntimeMode(RuntimeExecutionMode.BATCH);
// fromElements -> Map -> Print
DataStream<Integer> sourceDataStream = env.fromElements(1, 2, 3);
@@ -1367,6 +1369,30 @@ class StreamingJobGraphGeneratorTest {
.isCloseTo(expectedBatchFrac, Offset.offset(delta));
}
+ @Test
+ void testHybridShuffleModeInNonBatchMode() {
+ Configuration configuration = new Configuration();
+ // set all edge to HYBRID_FULL result partition type.
+ configuration.set(
+ ExecutionOptions.BATCH_SHUFFLE_MODE, BatchShuffleMode.ALL_EXCHANGES_HYBRID_FULL);
+ configuration.set(ExecutionOptions.RUNTIME_MODE, RuntimeExecutionMode.STREAMING);
+ StreamExecutionEnvironment env =
+ StreamExecutionEnvironment.getExecutionEnvironment(configuration);
+ env.disableOperatorChaining();
+ DataStreamSource<Integer> source = env.fromElements(1, 2, 3);
+ final DataStream<Integer> partitioned =
+ new DataStream<>(
+ env,
+ new PartitionTransformation<>(
+ source.getTransformation(),
+ new RebalancePartitioner<>(),
+ StreamExchangeMode.HYBRID_FULL));
+ partitioned.addSink(new DiscardingSink<>());
+ StreamGraph streamGraph = env.getStreamGraph();
+ assertThatThrownBy(() -> StreamingJobGraphGenerator.createJobGraph(streamGraph))
+ .isInstanceOf(IllegalStateException.class);
+ }
+
@Test
void testSetNonDefaultSlotSharingInHybridMode() {
Configuration configuration = new Configuration();