You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tw...@apache.org on 2021/07/21 13:41:30 UTC
[flink] 03/04: [FLINK-23402][streaming-java] Refactor
GlobalDataExchangeMode to GlobalStreamExchangeMode
This is an automated email from the ASF dual-hosted git repository.
twalthr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 156f517d387202ac292bde5bfac423a23908b7a2
Author: Timo Walther <tw...@apache.org>
AuthorDate: Mon Jul 19 16:14:51 2021 +0200
[FLINK-23402][streaming-java] Refactor GlobalDataExchangeMode to GlobalStreamExchangeMode
---
.../apache/flink/table/tpcds/TpcdsTestProgram.java | 4 +-
.../environment/StreamExecutionEnvironment.java | 5 ++-
...angeMode.java => GlobalStreamExchangeMode.java} | 2 +-
.../flink/streaming/api/graph/StreamGraph.java | 10 ++---
.../streaming/api/graph/StreamGraphGenerator.java | 8 ++--
.../api/graph/StreamingJobGraphGenerator.java | 4 +-
...amGraphGeneratorExecutionModeDetectionTest.java | 25 ++++++-----
...GeneratorWithGlobalStreamExchangeModeTest.java} | 22 +++++-----
.../plan/nodes/exec/batch/BatchExecExchange.java | 4 +-
.../flink/table/planner/utils/ExecutorUtils.java | 8 ++--
.../planner/utils/StreamExchangeModeUtils.java | 10 ++---
.../physical/batch/BatchPhysicalExchange.scala | 4 +-
.../planner/utils/StreamExchangeModeUtilsTest.java | 48 +++++++++++-----------
.../planner/runtime/utils/BatchTestBase.scala | 4 +-
.../flink/table/planner/utils/TableTestBase.scala | 4 +-
.../flink/test/runtime/BlockingShuffleITCase.java | 4 +-
16 files changed, 85 insertions(+), 81 deletions(-)
diff --git a/flink-end-to-end-tests/flink-tpcds-test/src/main/java/org/apache/flink/table/tpcds/TpcdsTestProgram.java b/flink-end-to-end-tests/flink-tpcds-test/src/main/java/org/apache/flink/table/tpcds/TpcdsTestProgram.java
index 1ade3bb..3f46dc2 100644
--- a/flink-end-to-end-tests/flink-tpcds-test/src/main/java/org/apache/flink/table/tpcds/TpcdsTestProgram.java
+++ b/flink-end-to-end-tests/flink-tpcds-test/src/main/java/org/apache/flink/table/tpcds/TpcdsTestProgram.java
@@ -20,7 +20,7 @@ package org.apache.flink.table.tpcds;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.core.fs.FileSystem;
-import org.apache.flink.streaming.api.graph.GlobalDataExchangeMode;
+import org.apache.flink.streaming.api.graph.GlobalStreamExchangeMode;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
@@ -145,7 +145,7 @@ public class TpcdsTestProgram {
.getConfiguration()
.setString(
ExecutionConfigOptions.TABLE_EXEC_SHUFFLE_MODE,
- GlobalDataExchangeMode.POINTWISE_EDGES_PIPELINED.toString());
+ GlobalStreamExchangeMode.POINTWISE_EDGES_PIPELINED.toString());
tEnv.getConfig()
.getConfiguration()
.setLong(
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
index 9670b39..8b0dd6a 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
@@ -93,7 +93,7 @@ import org.apache.flink.streaming.api.functions.source.SocketTextStreamFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.functions.source.StatefulSequenceSource;
import org.apache.flink.streaming.api.functions.source.TimestampedFileInputSplit;
-import org.apache.flink.streaming.api.graph.GlobalDataExchangeMode;
+import org.apache.flink.streaming.api.graph.GlobalStreamExchangeMode;
import org.apache.flink.streaming.api.graph.StreamGraph;
import org.apache.flink.streaming.api.graph.StreamGraphGenerator;
import org.apache.flink.streaming.api.graph.StreamingJobGraphGenerator;
@@ -2110,7 +2110,8 @@ public class StreamExecutionEnvironment {
if (configuration.get(ExecutionOptions.RUNTIME_MODE) == RuntimeExecutionMode.BATCH
&& streamGraph.hasFineGrainedResource()) {
if (configuration.get(ClusterOptions.FINE_GRAINED_SHUFFLE_MODE_ALL_BLOCKING)) {
- streamGraph.setGlobalDataExchangeMode(GlobalDataExchangeMode.ALL_EDGES_BLOCKING);
+ streamGraph.setGlobalStreamExchangeMode(
+ GlobalStreamExchangeMode.ALL_EDGES_BLOCKING);
} else {
throw new IllegalConfigurationException(
"At the moment, fine-grained resource management requires batch workloads to "
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/GlobalDataExchangeMode.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/GlobalStreamExchangeMode.java
similarity index 98%
rename from flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/GlobalDataExchangeMode.java
rename to flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/GlobalStreamExchangeMode.java
index 3118f7f..2befa4b 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/GlobalDataExchangeMode.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/GlobalStreamExchangeMode.java
@@ -27,7 +27,7 @@ import org.apache.flink.streaming.runtime.partitioner.RescalePartitioner;
* This mode decides the default {@link ResultPartitionType} of job edges. Note that this only
* affects job edges which are {@link StreamExchangeMode#UNDEFINED}.
*/
-public enum GlobalDataExchangeMode {
+public enum GlobalStreamExchangeMode {
/** Set all job edges to be {@link ResultPartitionType#BLOCKING}. */
ALL_EDGES_BLOCKING,
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java
index a641898..df468c4 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java
@@ -104,7 +104,7 @@ public class StreamGraph implements Pipeline {
private TimeCharacteristic timeCharacteristic;
- private GlobalDataExchangeMode globalDataExchangeMode;
+ private GlobalStreamExchangeMode globalExchangeMode;
/** Flag to indicate whether to put all vertices into the same slot sharing group by default. */
private boolean allVerticesInSameSlotSharingGroupByDefault = true;
@@ -237,12 +237,12 @@ public class StreamGraph implements Pipeline {
this.timeCharacteristic = timeCharacteristic;
}
- public GlobalDataExchangeMode getGlobalDataExchangeMode() {
- return globalDataExchangeMode;
+ public GlobalStreamExchangeMode getGlobalStreamExchangeMode() {
+ return globalExchangeMode;
}
- public void setGlobalDataExchangeMode(GlobalDataExchangeMode globalDataExchangeMode) {
- this.globalDataExchangeMode = globalDataExchangeMode;
+ public void setGlobalStreamExchangeMode(GlobalStreamExchangeMode globalExchangeMode) {
+ this.globalExchangeMode = globalExchangeMode;
}
public void setSlotSharingGroupResource(
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 d1b209c..72c7d37 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
@@ -362,7 +362,7 @@ public class StreamGraphGenerator {
}
graph.setAllVerticesInSameSlotSharingGroupByDefault(false);
- graph.setGlobalDataExchangeMode(GlobalDataExchangeMode.FORWARD_EDGES_PIPELINED);
+ graph.setGlobalStreamExchangeMode(GlobalStreamExchangeMode.FORWARD_EDGES_PIPELINED);
setDefaultBufferTimeout(-1);
setBatchStateBackendAndTimerService(graph);
} else {
@@ -373,10 +373,10 @@ public class StreamGraphGenerator {
if (checkpointConfig.isApproximateLocalRecoveryEnabled()) {
checkApproximateLocalRecoveryCompatibility();
- graph.setGlobalDataExchangeMode(
- GlobalDataExchangeMode.ALL_EDGES_PIPELINED_APPROXIMATE);
+ graph.setGlobalStreamExchangeMode(
+ GlobalStreamExchangeMode.ALL_EDGES_PIPELINED_APPROXIMATE);
} else {
- graph.setGlobalDataExchangeMode(GlobalDataExchangeMode.ALL_EDGES_PIPELINED);
+ graph.setGlobalStreamExchangeMode(GlobalStreamExchangeMode.ALL_EDGES_PIPELINED);
}
}
}
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 a4c0936..6f8fa5b 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
@@ -838,7 +838,7 @@ public class StreamingJobGraphGenerator {
}
private ResultPartitionType determineResultPartitionType(StreamPartitioner<?> partitioner) {
- switch (streamGraph.getGlobalDataExchangeMode()) {
+ switch (streamGraph.getGlobalStreamExchangeMode()) {
case ALL_EDGES_BLOCKING:
return ResultPartitionType.BLOCKING;
case FORWARD_EDGES_PIPELINED:
@@ -860,7 +860,7 @@ public class StreamingJobGraphGenerator {
default:
throw new RuntimeException(
"Unrecognized global data exchange mode "
- + streamGraph.getGlobalDataExchangeMode());
+ + streamGraph.getGlobalStreamExchangeMode());
}
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamGraphGeneratorExecutionModeDetectionTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamGraphGeneratorExecutionModeDetectionTest.java
index cb83685..4e96ec7 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamGraphGeneratorExecutionModeDetectionTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamGraphGeneratorExecutionModeDetectionTest.java
@@ -76,7 +76,10 @@ public class StreamGraphGeneratorExecutionModeDetectionTest extends TestLogger {
assertThat(
environment.getStreamGraph(),
hasProperties(
- GlobalDataExchangeMode.ALL_EDGES_PIPELINED, JobType.STREAMING, true, true));
+ GlobalStreamExchangeMode.ALL_EDGES_PIPELINED,
+ JobType.STREAMING,
+ true,
+ true));
}
@Test
@@ -94,7 +97,7 @@ public class StreamGraphGeneratorExecutionModeDetectionTest extends TestLogger {
assertThat(
environment.getStreamGraph(),
hasProperties(
- GlobalDataExchangeMode.ALL_EDGES_PIPELINED,
+ GlobalStreamExchangeMode.ALL_EDGES_PIPELINED,
JobType.STREAMING,
false,
true));
@@ -124,7 +127,7 @@ public class StreamGraphGeneratorExecutionModeDetectionTest extends TestLogger {
assertThat(
streamGraph,
hasProperties(
- GlobalDataExchangeMode.FORWARD_EDGES_PIPELINED,
+ GlobalStreamExchangeMode.FORWARD_EDGES_PIPELINED,
JobType.BATCH,
false,
false));
@@ -177,7 +180,7 @@ public class StreamGraphGeneratorExecutionModeDetectionTest extends TestLogger {
assertThat(
graph,
hasProperties(
- GlobalDataExchangeMode.ALL_EDGES_PIPELINED,
+ GlobalStreamExchangeMode.ALL_EDGES_PIPELINED,
JobType.STREAMING,
false,
true));
@@ -193,7 +196,7 @@ public class StreamGraphGeneratorExecutionModeDetectionTest extends TestLogger {
assertThat(
graph,
hasProperties(
- GlobalDataExchangeMode.FORWARD_EDGES_PIPELINED,
+ GlobalStreamExchangeMode.FORWARD_EDGES_PIPELINED,
JobType.BATCH,
false,
false));
@@ -209,7 +212,7 @@ public class StreamGraphGeneratorExecutionModeDetectionTest extends TestLogger {
assertThat(
graph,
hasProperties(
- GlobalDataExchangeMode.ALL_EDGES_PIPELINED,
+ GlobalStreamExchangeMode.ALL_EDGES_PIPELINED,
JobType.STREAMING,
false,
true));
@@ -229,7 +232,7 @@ public class StreamGraphGeneratorExecutionModeDetectionTest extends TestLogger {
assertThat(
graph,
hasProperties(
- GlobalDataExchangeMode.ALL_EDGES_PIPELINED,
+ GlobalStreamExchangeMode.ALL_EDGES_PIPELINED,
JobType.STREAMING,
false,
true));
@@ -245,7 +248,7 @@ public class StreamGraphGeneratorExecutionModeDetectionTest extends TestLogger {
assertThat(
graph,
hasProperties(
- GlobalDataExchangeMode.FORWARD_EDGES_PIPELINED,
+ GlobalStreamExchangeMode.FORWARD_EDGES_PIPELINED,
JobType.BATCH,
false,
false));
@@ -255,7 +258,7 @@ public class StreamGraphGeneratorExecutionModeDetectionTest extends TestLogger {
assertThat(
streamingGraph,
hasProperties(
- GlobalDataExchangeMode.ALL_EDGES_PIPELINED,
+ GlobalStreamExchangeMode.ALL_EDGES_PIPELINED,
JobType.STREAMING,
false,
true));
@@ -285,7 +288,7 @@ public class StreamGraphGeneratorExecutionModeDetectionTest extends TestLogger {
}
private static TypeSafeMatcher<StreamGraph> hasProperties(
- final GlobalDataExchangeMode exchangeMode,
+ final GlobalStreamExchangeMode exchangeMode,
final JobType jobType,
final boolean isCheckpointingEnabled,
final boolean isAllVerticesInSameSlotSharingGroupByDefault) {
@@ -293,7 +296,7 @@ public class StreamGraphGeneratorExecutionModeDetectionTest extends TestLogger {
return new TypeSafeMatcher<StreamGraph>() {
@Override
protected boolean matchesSafely(StreamGraph actualStreamGraph) {
- return exchangeMode == actualStreamGraph.getGlobalDataExchangeMode()
+ return exchangeMode == actualStreamGraph.getGlobalStreamExchangeMode()
&& jobType == actualStreamGraph.getJobType()
&& actualStreamGraph.getCheckpointConfig().isCheckpointingEnabled()
== isCheckpointingEnabled
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGeneratorWithGlobalDataExchangeModeTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGeneratorWithGlobalStreamExchangeModeTest.java
similarity index 90%
rename from flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGeneratorWithGlobalDataExchangeModeTest.java
rename to flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGeneratorWithGlobalStreamExchangeModeTest.java
index e1b6dd1..4042d84 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGeneratorWithGlobalDataExchangeModeTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGeneratorWithGlobalStreamExchangeModeTest.java
@@ -37,23 +37,23 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertThat;
/**
- * Tests for {@link StreamingJobGraphGenerator} on different {@link GlobalDataExchangeMode}
+ * Tests for {@link StreamingJobGraphGenerator} on different {@link GlobalStreamExchangeMode}
* settings.
*/
-public class StreamingJobGraphGeneratorWithGlobalDataExchangeModeTest extends TestLogger {
+public class StreamingJobGraphGeneratorWithGlobalStreamExchangeModeTest extends TestLogger {
@Test
- public void testDefaultGlobalDataExchangeModeIsAllEdgesPipelined() {
+ public void testDefaultGlobalExchangeModeIsAllEdgesPipelined() {
final StreamGraph streamGraph = createStreamGraph();
assertThat(
- streamGraph.getGlobalDataExchangeMode(),
- is(GlobalDataExchangeMode.ALL_EDGES_PIPELINED));
+ streamGraph.getGlobalStreamExchangeMode(),
+ is(GlobalStreamExchangeMode.ALL_EDGES_PIPELINED));
}
@Test
public void testAllEdgesBlockingMode() {
final StreamGraph streamGraph = createStreamGraph();
- streamGraph.setGlobalDataExchangeMode(GlobalDataExchangeMode.ALL_EDGES_BLOCKING);
+ streamGraph.setGlobalStreamExchangeMode(GlobalStreamExchangeMode.ALL_EDGES_BLOCKING);
final JobGraph jobGraph = StreamingJobGraphGenerator.createJobGraph(streamGraph);
final List<JobVertex> verticesSorted = jobGraph.getVerticesSortedTopologicallyFromSources();
@@ -75,7 +75,7 @@ public class StreamingJobGraphGeneratorWithGlobalDataExchangeModeTest extends Te
@Test
public void testAllEdgesPipelinedMode() {
final StreamGraph streamGraph = createStreamGraph();
- streamGraph.setGlobalDataExchangeMode(GlobalDataExchangeMode.ALL_EDGES_PIPELINED);
+ streamGraph.setGlobalStreamExchangeMode(GlobalStreamExchangeMode.ALL_EDGES_PIPELINED);
final JobGraph jobGraph = StreamingJobGraphGenerator.createJobGraph(streamGraph);
final List<JobVertex> verticesSorted = jobGraph.getVerticesSortedTopologicallyFromSources();
@@ -97,7 +97,7 @@ public class StreamingJobGraphGeneratorWithGlobalDataExchangeModeTest extends Te
@Test
public void testForwardEdgesPipelinedMode() {
final StreamGraph streamGraph = createStreamGraph();
- streamGraph.setGlobalDataExchangeMode(GlobalDataExchangeMode.FORWARD_EDGES_PIPELINED);
+ streamGraph.setGlobalStreamExchangeMode(GlobalStreamExchangeMode.FORWARD_EDGES_PIPELINED);
final JobGraph jobGraph = StreamingJobGraphGenerator.createJobGraph(streamGraph);
final List<JobVertex> verticesSorted = jobGraph.getVerticesSortedTopologicallyFromSources();
@@ -119,7 +119,7 @@ public class StreamingJobGraphGeneratorWithGlobalDataExchangeModeTest extends Te
@Test
public void testPointwiseEdgesPipelinedMode() {
final StreamGraph streamGraph = createStreamGraph();
- streamGraph.setGlobalDataExchangeMode(GlobalDataExchangeMode.POINTWISE_EDGES_PIPELINED);
+ streamGraph.setGlobalStreamExchangeMode(GlobalStreamExchangeMode.POINTWISE_EDGES_PIPELINED);
final JobGraph jobGraph = StreamingJobGraphGenerator.createJobGraph(streamGraph);
final List<JobVertex> verticesSorted = jobGraph.getVerticesSortedTopologicallyFromSources();
@@ -139,7 +139,7 @@ public class StreamingJobGraphGeneratorWithGlobalDataExchangeModeTest extends Te
}
@Test
- public void testGlobalDataExchangeModeDoesNotOverrideSpecifiedExchangeMode() {
+ public void testGlobalExchangeModeDoesNotOverrideSpecifiedExchangeMode() {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
final DataStream<Integer> source = env.fromElements(1, 2, 3).setParallelism(1);
final DataStream<Integer> forward =
@@ -151,7 +151,7 @@ public class StreamingJobGraphGeneratorWithGlobalDataExchangeModeTest extends Te
StreamExchangeMode.PIPELINED));
forward.map(i -> i).startNewChain().setParallelism(1);
final StreamGraph streamGraph = env.getStreamGraph();
- streamGraph.setGlobalDataExchangeMode(GlobalDataExchangeMode.ALL_EDGES_BLOCKING);
+ streamGraph.setGlobalStreamExchangeMode(GlobalStreamExchangeMode.ALL_EDGES_BLOCKING);
final JobGraph jobGraph = StreamingJobGraphGenerator.createJobGraph(streamGraph);
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecExchange.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecExchange.java
index c4b9925..08e7542 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecExchange.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecExchange.java
@@ -22,7 +22,7 @@ import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.dag.Transformation;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.streaming.api.graph.GlobalDataExchangeMode;
+import org.apache.flink.streaming.api.graph.GlobalStreamExchangeMode;
import org.apache.flink.streaming.api.transformations.PartitionTransformation;
import org.apache.flink.streaming.api.transformations.StreamExchangeMode;
import org.apache.flink.streaming.runtime.partitioner.BroadcastPartitioner;
@@ -156,7 +156,7 @@ public class BatchExecExchange extends CommonExecExchange implements BatchExecNo
return StreamExchangeMode.BATCH;
}
if (config.getString(ExecutionConfigOptions.TABLE_EXEC_SHUFFLE_MODE)
- .equalsIgnoreCase(GlobalDataExchangeMode.ALL_EDGES_BLOCKING.toString())) {
+ .equalsIgnoreCase(GlobalStreamExchangeMode.ALL_EDGES_BLOCKING.toString())) {
return StreamExchangeMode.BATCH;
} else {
return StreamExchangeMode.UNDEFINED;
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/utils/ExecutorUtils.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/utils/ExecutorUtils.java
index 4871560..5ed9b30 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/utils/ExecutorUtils.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/utils/ExecutorUtils.java
@@ -23,7 +23,7 @@ import org.apache.flink.api.common.operators.ResourceSpec;
import org.apache.flink.api.dag.Transformation;
import org.apache.flink.runtime.jobgraph.JobType;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.streaming.api.graph.GlobalDataExchangeMode;
+import org.apache.flink.streaming.api.graph.GlobalStreamExchangeMode;
import org.apache.flink.streaming.api.graph.StreamGraph;
import org.apache.flink.streaming.api.graph.StreamGraphGenerator;
import org.apache.flink.table.api.TableConfig;
@@ -80,11 +80,11 @@ public class ExecutorUtils {
if (streamGraph.getCheckpointConfig().isCheckpointingEnabled()) {
throw new IllegalArgumentException("Checkpoint is not supported for batch jobs.");
}
- streamGraph.setGlobalDataExchangeMode(getGlobalDataExchangeMode(tableConfig));
+ streamGraph.setGlobalStreamExchangeMode(getGlobalStreamExchangeMode(tableConfig));
}
- private static GlobalDataExchangeMode getGlobalDataExchangeMode(TableConfig tableConfig) {
- return StreamExchangeModeUtils.getShuffleModeAsGlobalDataExchangeMode(
+ private static GlobalStreamExchangeMode getGlobalStreamExchangeMode(TableConfig tableConfig) {
+ return StreamExchangeModeUtils.getShuffleModeAsGlobalStreamExchangeMode(
tableConfig.getConfiguration());
}
}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/utils/StreamExchangeModeUtils.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/utils/StreamExchangeModeUtils.java
index 3ecf718..e31dbc4 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/utils/StreamExchangeModeUtils.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/utils/StreamExchangeModeUtils.java
@@ -19,7 +19,7 @@
package org.apache.flink.table.planner.utils;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.streaming.api.graph.GlobalDataExchangeMode;
+import org.apache.flink.streaming.api.graph.GlobalStreamExchangeMode;
import org.apache.flink.table.api.config.ExecutionConfigOptions;
/** Utility class to load job-wide exchange mode. */
@@ -29,13 +29,13 @@ public class StreamExchangeModeUtils {
static final String ALL_EDGES_PIPELINED_LEGACY = "pipelined";
- static GlobalDataExchangeMode getShuffleModeAsGlobalDataExchangeMode(
+ static GlobalStreamExchangeMode getShuffleModeAsGlobalStreamExchangeMode(
final Configuration configuration) {
final String value =
configuration.getString(ExecutionConfigOptions.TABLE_EXEC_SHUFFLE_MODE);
try {
- return GlobalDataExchangeMode.valueOf(convertLegacyShuffleMode(value).toUpperCase());
+ return GlobalStreamExchangeMode.valueOf(convertLegacyShuffleMode(value).toUpperCase());
} catch (IllegalArgumentException e) {
throw new IllegalArgumentException(
String.format(
@@ -47,9 +47,9 @@ public class StreamExchangeModeUtils {
private static String convertLegacyShuffleMode(final String shuffleMode) {
switch (shuffleMode.toLowerCase()) {
case ALL_EDGES_BLOCKING_LEGACY:
- return GlobalDataExchangeMode.ALL_EDGES_BLOCKING.toString();
+ return GlobalStreamExchangeMode.ALL_EDGES_BLOCKING.toString();
case ALL_EDGES_PIPELINED_LEGACY:
- return GlobalDataExchangeMode.ALL_EDGES_PIPELINED.toString();
+ return GlobalStreamExchangeMode.ALL_EDGES_PIPELINED.toString();
default:
return shuffleMode;
}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalExchange.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalExchange.scala
index fc00199..50571ad 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalExchange.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalExchange.scala
@@ -18,7 +18,7 @@
package org.apache.flink.table.planner.plan.nodes.physical.batch
-import org.apache.flink.streaming.api.graph.GlobalDataExchangeMode
+import org.apache.flink.streaming.api.graph.GlobalStreamExchangeMode
import org.apache.flink.streaming.api.transformations.StreamExchangeMode
import org.apache.flink.table.api.config.ExecutionConfigOptions
import org.apache.flink.table.planner.calcite.FlinkTypeFactory
@@ -75,7 +75,7 @@ class BatchPhysicalExchange(
private def getExchangeMode: StreamExchangeMode = {
val tableConfig = FlinkRelOptUtil.getTableConfigFromContext(this)
if (tableConfig.getConfiguration.getString(ExecutionConfigOptions.TABLE_EXEC_SHUFFLE_MODE)
- .equalsIgnoreCase(GlobalDataExchangeMode.ALL_EDGES_BLOCKING.toString)) {
+ .equalsIgnoreCase(GlobalStreamExchangeMode.ALL_EDGES_BLOCKING.toString)) {
StreamExchangeMode.BATCH
} else {
StreamExchangeMode.UNDEFINED
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/utils/StreamExchangeModeUtilsTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/utils/StreamExchangeModeUtilsTest.java
index c0078da..46fb9f6 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/utils/StreamExchangeModeUtilsTest.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/utils/StreamExchangeModeUtilsTest.java
@@ -19,7 +19,7 @@
package org.apache.flink.table.planner.utils;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.streaming.api.graph.GlobalDataExchangeMode;
+import org.apache.flink.streaming.api.graph.GlobalStreamExchangeMode;
import org.apache.flink.table.api.config.ExecutionConfigOptions;
import org.apache.flink.util.TestLogger;
@@ -36,31 +36,31 @@ public class StreamExchangeModeUtilsTest extends TestLogger {
configuration.setString(
ExecutionConfigOptions.TABLE_EXEC_SHUFFLE_MODE,
- GlobalDataExchangeMode.ALL_EDGES_BLOCKING.toString());
+ GlobalStreamExchangeMode.ALL_EDGES_BLOCKING.toString());
assertEquals(
- GlobalDataExchangeMode.ALL_EDGES_BLOCKING,
- StreamExchangeModeUtils.getShuffleModeAsGlobalDataExchangeMode(configuration));
+ GlobalStreamExchangeMode.ALL_EDGES_BLOCKING,
+ StreamExchangeModeUtils.getShuffleModeAsGlobalStreamExchangeMode(configuration));
configuration.setString(
ExecutionConfigOptions.TABLE_EXEC_SHUFFLE_MODE,
- GlobalDataExchangeMode.FORWARD_EDGES_PIPELINED.toString());
+ GlobalStreamExchangeMode.FORWARD_EDGES_PIPELINED.toString());
assertEquals(
- GlobalDataExchangeMode.FORWARD_EDGES_PIPELINED,
- StreamExchangeModeUtils.getShuffleModeAsGlobalDataExchangeMode(configuration));
+ GlobalStreamExchangeMode.FORWARD_EDGES_PIPELINED,
+ StreamExchangeModeUtils.getShuffleModeAsGlobalStreamExchangeMode(configuration));
configuration.setString(
ExecutionConfigOptions.TABLE_EXEC_SHUFFLE_MODE,
- GlobalDataExchangeMode.POINTWISE_EDGES_PIPELINED.toString());
+ GlobalStreamExchangeMode.POINTWISE_EDGES_PIPELINED.toString());
assertEquals(
- GlobalDataExchangeMode.POINTWISE_EDGES_PIPELINED,
- StreamExchangeModeUtils.getShuffleModeAsGlobalDataExchangeMode(configuration));
+ GlobalStreamExchangeMode.POINTWISE_EDGES_PIPELINED,
+ StreamExchangeModeUtils.getShuffleModeAsGlobalStreamExchangeMode(configuration));
configuration.setString(
ExecutionConfigOptions.TABLE_EXEC_SHUFFLE_MODE,
- GlobalDataExchangeMode.ALL_EDGES_PIPELINED.toString());
+ GlobalStreamExchangeMode.ALL_EDGES_PIPELINED.toString());
assertEquals(
- GlobalDataExchangeMode.ALL_EDGES_PIPELINED,
- StreamExchangeModeUtils.getShuffleModeAsGlobalDataExchangeMode(configuration));
+ GlobalStreamExchangeMode.ALL_EDGES_PIPELINED,
+ StreamExchangeModeUtils.getShuffleModeAsGlobalStreamExchangeMode(configuration));
}
@Test
@@ -71,15 +71,15 @@ public class StreamExchangeModeUtilsTest extends TestLogger {
ExecutionConfigOptions.TABLE_EXEC_SHUFFLE_MODE,
StreamExchangeModeUtils.ALL_EDGES_BLOCKING_LEGACY);
assertEquals(
- GlobalDataExchangeMode.ALL_EDGES_BLOCKING,
- StreamExchangeModeUtils.getShuffleModeAsGlobalDataExchangeMode(configuration));
+ GlobalStreamExchangeMode.ALL_EDGES_BLOCKING,
+ StreamExchangeModeUtils.getShuffleModeAsGlobalStreamExchangeMode(configuration));
configuration.setString(
ExecutionConfigOptions.TABLE_EXEC_SHUFFLE_MODE,
StreamExchangeModeUtils.ALL_EDGES_PIPELINED_LEGACY);
assertEquals(
- GlobalDataExchangeMode.ALL_EDGES_PIPELINED,
- StreamExchangeModeUtils.getShuffleModeAsGlobalDataExchangeMode(configuration));
+ GlobalStreamExchangeMode.ALL_EDGES_PIPELINED,
+ StreamExchangeModeUtils.getShuffleModeAsGlobalStreamExchangeMode(configuration));
}
@Test
@@ -89,27 +89,27 @@ public class StreamExchangeModeUtilsTest extends TestLogger {
configuration.setString(
ExecutionConfigOptions.TABLE_EXEC_SHUFFLE_MODE, "Forward_edges_PIPELINED");
assertEquals(
- GlobalDataExchangeMode.FORWARD_EDGES_PIPELINED,
- StreamExchangeModeUtils.getShuffleModeAsGlobalDataExchangeMode(configuration));
+ GlobalStreamExchangeMode.FORWARD_EDGES_PIPELINED,
+ StreamExchangeModeUtils.getShuffleModeAsGlobalStreamExchangeMode(configuration));
configuration.setString(ExecutionConfigOptions.TABLE_EXEC_SHUFFLE_MODE, "Pipelined");
assertEquals(
- GlobalDataExchangeMode.ALL_EDGES_PIPELINED,
- StreamExchangeModeUtils.getShuffleModeAsGlobalDataExchangeMode(configuration));
+ GlobalStreamExchangeMode.ALL_EDGES_PIPELINED,
+ StreamExchangeModeUtils.getShuffleModeAsGlobalStreamExchangeMode(configuration));
}
@Test
public void testGetDefaultShuffleMode() {
final Configuration configuration = new Configuration();
assertEquals(
- GlobalDataExchangeMode.ALL_EDGES_BLOCKING,
- StreamExchangeModeUtils.getShuffleModeAsGlobalDataExchangeMode(configuration));
+ GlobalStreamExchangeMode.ALL_EDGES_BLOCKING,
+ StreamExchangeModeUtils.getShuffleModeAsGlobalStreamExchangeMode(configuration));
}
@Test(expected = IllegalArgumentException.class)
public void testGetInvalidShuffleMode() {
final Configuration configuration = new Configuration();
configuration.setString(ExecutionConfigOptions.TABLE_EXEC_SHUFFLE_MODE, "invalid-value");
- StreamExchangeModeUtils.getShuffleModeAsGlobalDataExchangeMode(configuration);
+ StreamExchangeModeUtils.getShuffleModeAsGlobalStreamExchangeMode(configuration);
}
}
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/utils/BatchTestBase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/utils/BatchTestBase.scala
index d1e1eef..fa4dc30 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/utils/BatchTestBase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/utils/BatchTestBase.scala
@@ -22,7 +22,7 @@ import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.streaming.api.datastream.DataStream
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
-import org.apache.flink.streaming.api.graph.GlobalDataExchangeMode
+import org.apache.flink.streaming.api.graph.GlobalStreamExchangeMode
import org.apache.flink.table.api._
import org.apache.flink.table.api.config.ExecutionConfigOptions._
import org.apache.flink.table.api.internal.TableEnvironmentImpl
@@ -521,6 +521,6 @@ object BatchTestBase {
conf.getConfiguration.setInteger(TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM, DEFAULT_PARALLELISM)
conf.getConfiguration.setString(
TABLE_EXEC_SHUFFLE_MODE,
- GlobalDataExchangeMode.ALL_EDGES_PIPELINED.toString)
+ GlobalStreamExchangeMode.ALL_EDGES_PIPELINED.toString)
}
}
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/utils/TableTestBase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/utils/TableTestBase.scala
index 3659d81..d1f4107 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/utils/TableTestBase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/utils/TableTestBase.scala
@@ -22,7 +22,7 @@ import org.apache.flink.api.java.typeutils.{PojoTypeInfo, RowTypeInfo, TupleType
import org.apache.flink.api.scala.typeutils.CaseClassTypeInfo
import org.apache.flink.streaming.api.datastream.DataStream
import org.apache.flink.streaming.api.environment.{LocalStreamEnvironment, StreamExecutionEnvironment}
-import org.apache.flink.streaming.api.graph.GlobalDataExchangeMode
+import org.apache.flink.streaming.api.graph.GlobalStreamExchangeMode
import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment => ScalaStreamExecEnv}
import org.apache.flink.streaming.api.{TimeCharacteristic, environment}
import org.apache.flink.table.api._
@@ -997,7 +997,7 @@ abstract class TableTestUtil(
val tableEnv: TableEnvironment = testingTableEnv
tableEnv.getConfig.getConfiguration.setString(
ExecutionConfigOptions.TABLE_EXEC_SHUFFLE_MODE,
- GlobalDataExchangeMode.ALL_EDGES_PIPELINED.toString)
+ GlobalStreamExchangeMode.ALL_EDGES_PIPELINED.toString)
private val env: StreamExecutionEnvironment = getPlanner.getExecEnv
diff --git a/flink-tests/src/test/java/org/apache/flink/test/runtime/BlockingShuffleITCase.java b/flink-tests/src/test/java/org/apache/flink/test/runtime/BlockingShuffleITCase.java
index e2eaa33..0b38c5b 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/runtime/BlockingShuffleITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/runtime/BlockingShuffleITCase.java
@@ -27,7 +27,7 @@ import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;
-import org.apache.flink.streaming.api.graph.GlobalDataExchangeMode;
+import org.apache.flink.streaming.api.graph.GlobalStreamExchangeMode;
import org.apache.flink.streaming.api.graph.StreamGraph;
import org.apache.flink.streaming.api.graph.StreamingJobGraphGenerator;
@@ -92,7 +92,7 @@ public class BlockingShuffleITCase {
.addSink(new VerifySink());
StreamGraph streamGraph = env.getStreamGraph();
- streamGraph.setGlobalDataExchangeMode(GlobalDataExchangeMode.ALL_EDGES_BLOCKING);
+ streamGraph.setGlobalStreamExchangeMode(GlobalStreamExchangeMode.ALL_EDGES_BLOCKING);
// a scheduler supporting batch jobs is required for this job graph, because it contains
// blocking data exchanges.
// The scheduler is selected based on the JobType.