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:29 UTC
[flink] 02/04: [FLINK-23402][streaming-java] Fix minor code issues
around 'shuffle mode'
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 86f54c89c7866647e50d3957026bd0d28869ea8d
Author: Timo Walther <tw...@apache.org>
AuthorDate: Mon Jul 19 15:02:38 2021 +0200
[FLINK-23402][streaming-java] Fix minor code issues around 'shuffle mode'
---
.../flink/streaming/api/graph/StreamEdge.java | 2 ++
.../transformations/PartitionTransformation.java | 20 ++++++------
.../api/transformations/StreamExchangeMode.java | 9 +++---
.../api/graph/StreamingJobGraphGeneratorTest.java | 36 +++++++++++-----------
...aphGeneratorWithGlobalDataExchangeModeTest.java | 2 +-
.../plan/nodes/exec/batch/BatchExecExchange.java | 2 +-
.../utils/InputPriorityConflictResolver.java | 2 +-
.../planner/utils/StreamExchangeModeUtils.java | 2 +-
8 files changed, 39 insertions(+), 36 deletions(-)
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 5ca627c..84c4110 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
@@ -194,6 +194,8 @@ public class StreamEdge implements Serializable {
+ typeNumber
+ ", outputPartitioner="
+ outputPartitioner
+ + ", exchangeMode="
+ + exchangeMode
+ ", bufferTimeout="
+ bufferTimeout
+ ", outputTag="
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/PartitionTransformation.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/PartitionTransformation.java
index 2521539..bc334d1 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/PartitionTransformation.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/PartitionTransformation.java
@@ -35,7 +35,7 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
* <p>This does not create a physical operation, it only affects how upstream operations are
* connected to downstream operations.
*
- * @param <T> The type of the elements that result from this {@code PartitionTransformation}
+ * @param <T> The type of the elements that result from this {@link PartitionTransformation}
*/
@Internal
public class PartitionTransformation<T> extends Transformation<T> {
@@ -47,23 +47,23 @@ public class PartitionTransformation<T> extends Transformation<T> {
private final StreamExchangeMode exchangeMode;
/**
- * Creates a new {@code PartitionTransformation} from the given input and {@link
+ * Creates a new {@link PartitionTransformation} from the given input and {@link
* StreamPartitioner}.
*
- * @param input The input {@code Transformation}
- * @param partitioner The {@code StreamPartitioner}
+ * @param input The input {@link Transformation}
+ * @param partitioner The {@link StreamPartitioner}
*/
public PartitionTransformation(Transformation<T> input, StreamPartitioner<T> partitioner) {
this(input, partitioner, StreamExchangeMode.UNDEFINED);
}
/**
- * Creates a new {@code PartitionTransformation} from the given input and {@link
+ * Creates a new {@link PartitionTransformation} from the given input and {@link
* StreamPartitioner}.
*
- * @param input The input {@code Transformation}
- * @param partitioner The {@code StreamPartitioner}
- * @param exchangeMode The {@code ShuffleMode}
+ * @param input The input {@link Transformation}
+ * @param partitioner The {@link StreamPartitioner}
+ * @param exchangeMode The {@link StreamExchangeMode}
*/
public PartitionTransformation(
Transformation<T> input,
@@ -76,8 +76,8 @@ public class PartitionTransformation<T> extends Transformation<T> {
}
/**
- * Returns the {@code StreamPartitioner} that must be used for partitioning the elements of the
- * input {@code Transformation}.
+ * Returns the {@link StreamPartitioner} that must be used for partitioning the elements of the
+ * input {@link Transformation}.
*/
public StreamPartitioner<T> getPartitioner() {
return partitioner;
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/StreamExchangeMode.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/StreamExchangeMode.java
index 4637c07..5752add 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/StreamExchangeMode.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/StreamExchangeMode.java
@@ -18,10 +18,11 @@
package org.apache.flink.streaming.api.transformations;
-import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.streaming.api.graph.StreamGraph;
-/** The shuffle mode defines the data exchange mode between operators. */
-@PublicEvolving
+/** The data exchange mode between operators during {@link StreamGraph} generation. */
+@Internal
public enum StreamExchangeMode {
/**
* Producer and consumer are online at the same time. Produced data is received by consumer
@@ -36,7 +37,7 @@ public enum StreamExchangeMode {
BATCH,
/**
- * The shuffle mode is undefined. It leaves it up to the framework to decide the shuffle mode.
+ * The exchange mode is undefined. It leaves it up to the framework to decide the exchange mode.
* The framework will pick one of {@link StreamExchangeMode#BATCH} or {@link
* StreamExchangeMode#PIPELINED} in the end.
*/
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 bfb82f3..23d2eb8 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
@@ -593,9 +593,9 @@ public class StreamingJobGraphGeneratorTest extends TestLogger {
assertTrue(operatorFactory instanceof SourceOperatorFactory);
}
- /** Test setting shuffle mode to {@link StreamExchangeMode#PIPELINED}. */
+ /** Test setting exchange mode to {@link StreamExchangeMode#PIPELINED}. */
@Test
- public void testShuffleModePipelined() {
+ public void testExchangeModePipelined() {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// fromElements -> Map -> Print
DataStream<Integer> sourceDataStream = env.fromElements(1, 2, 3);
@@ -624,18 +624,18 @@ public class StreamingJobGraphGeneratorTest extends TestLogger {
List<JobVertex> verticesSorted = jobGraph.getVerticesSortedTopologicallyFromSources();
assertEquals(2, verticesSorted.size());
- // it can be chained with PIPELINED shuffle mode
+ // it can be chained with PIPELINED exchange mode
JobVertex sourceAndMapVertex = verticesSorted.get(0);
- // PIPELINED shuffle mode is translated into PIPELINED_BOUNDED result partition
+ // PIPELINED exchange mode is translated into PIPELINED_BOUNDED result partition
assertEquals(
ResultPartitionType.PIPELINED_BOUNDED,
sourceAndMapVertex.getProducedDataSets().get(0).getResultType());
}
- /** Test setting shuffle mode to {@link StreamExchangeMode#BATCH}. */
+ /** Test setting exchange mode to {@link StreamExchangeMode#BATCH}. */
@Test
- public void testShuffleModeBatch() {
+ public void testExchangeModeBatch() {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// fromElements -> Map -> Print
DataStream<Integer> sourceDataStream = env.fromElements(1, 2, 3);
@@ -664,11 +664,11 @@ public class StreamingJobGraphGeneratorTest extends TestLogger {
List<JobVertex> verticesSorted = jobGraph.getVerticesSortedTopologicallyFromSources();
assertEquals(3, verticesSorted.size());
- // it can not be chained with BATCH shuffle mode
+ // it can not be chained with BATCH exchange mode
JobVertex sourceVertex = verticesSorted.get(0);
JobVertex mapVertex = verticesSorted.get(1);
- // BATCH shuffle mode is translated into BLOCKING result partition
+ // BATCH exchange mode is translated into BLOCKING result partition
assertEquals(
ResultPartitionType.BLOCKING,
sourceVertex.getProducedDataSets().get(0).getResultType());
@@ -677,9 +677,9 @@ public class StreamingJobGraphGeneratorTest extends TestLogger {
mapVertex.getProducedDataSets().get(0).getResultType());
}
- /** Test setting shuffle mode to {@link StreamExchangeMode#UNDEFINED}. */
+ /** Test setting exchange mode to {@link StreamExchangeMode#UNDEFINED}. */
@Test
- public void testShuffleModeUndefined() {
+ public void testExchangeModeUndefined() {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// fromElements -> Map -> Print
DataStream<Integer> sourceDataStream = env.fromElements(1, 2, 3);
@@ -708,10 +708,10 @@ public class StreamingJobGraphGeneratorTest extends TestLogger {
List<JobVertex> verticesSorted = jobGraph.getVerticesSortedTopologicallyFromSources();
assertEquals(2, verticesSorted.size());
- // it can be chained with UNDEFINED shuffle mode
+ // it can be chained with UNDEFINED exchange mode
JobVertex sourceAndMapVertex = verticesSorted.get(0);
- // UNDEFINED shuffle mode is translated into PIPELINED_BOUNDED result partition by default
+ // UNDEFINED exchange mode is translated into PIPELINED_BOUNDED result partition by default
assertEquals(
ResultPartitionType.PIPELINED_BOUNDED,
sourceAndMapVertex.getProducedDataSets().get(0).getResultType());
@@ -742,7 +742,7 @@ public class StreamingJobGraphGeneratorTest extends TestLogger {
env.disableOperatorChaining();
DataStream<Integer> source = env.fromElements(1);
source
- // set the same parallelism as the source to make it a FORWARD SHUFFLE
+ // set the same parallelism as the source to make it a FORWARD exchange
.map(value -> value)
.setParallelism(1)
.rescale()
@@ -783,16 +783,16 @@ public class StreamingJobGraphGeneratorTest extends TestLogger {
}
@Test(expected = UnsupportedOperationException.class)
- public void testConflictShuffleModeWithBufferTimeout() {
- testCompatibleShuffleModeWithBufferTimeout(StreamExchangeMode.BATCH);
+ public void testConflictExchangeModeWithBufferTimeout() {
+ testCompatibleExchangeModeWithBufferTimeout(StreamExchangeMode.BATCH);
}
@Test
- public void testNormalShuffleModeWithBufferTimeout() {
- testCompatibleShuffleModeWithBufferTimeout(StreamExchangeMode.PIPELINED);
+ public void testNormalExchangeModeWithBufferTimeout() {
+ testCompatibleExchangeModeWithBufferTimeout(StreamExchangeMode.PIPELINED);
}
- private void testCompatibleShuffleModeWithBufferTimeout(StreamExchangeMode exchangeMode) {
+ private void testCompatibleExchangeModeWithBufferTimeout(StreamExchangeMode exchangeMode) {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setBufferTimeout(100);
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/StreamingJobGraphGeneratorWithGlobalDataExchangeModeTest.java
index 7ee6462..e1b6dd1 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/StreamingJobGraphGeneratorWithGlobalDataExchangeModeTest.java
@@ -139,7 +139,7 @@ public class StreamingJobGraphGeneratorWithGlobalDataExchangeModeTest extends Te
}
@Test
- public void testGlobalDataExchangeModeDoesNotOverrideSpecifiedShuffleMode() {
+ public void testGlobalDataExchangeModeDoesNotOverrideSpecifiedExchangeMode() {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
final DataStream<Integer> source = env.fromElements(1, 2, 3).setParallelism(1);
final DataStream<Integer> forward =
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 3a5b3aa..c4b9925 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
@@ -56,7 +56,7 @@ import java.util.Optional;
* <p>TODO Remove this class once FLINK-21224 is finished.
*/
public class BatchExecExchange extends CommonExecExchange implements BatchExecNode<RowData> {
- // the required shuffle mode for reusable BatchExecExchange
+ // the required exchange mode for reusable BatchExecExchange
// if it's None, use value from configuration
@Nullable private StreamExchangeMode requiredExchangeMode;
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/processor/utils/InputPriorityConflictResolver.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/processor/utils/InputPriorityConflictResolver.java
index bc9a2c3..7721b7c 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/processor/utils/InputPriorityConflictResolver.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/processor/utils/InputPriorityConflictResolver.java
@@ -50,7 +50,7 @@ public class InputPriorityConflictResolver extends InputPriorityGraphGenerator {
* @param safeDamBehavior when checking for conflicts we'll ignore the edges with {@link
* InputProperty.DamBehavior} stricter or equal than this
* @param exchangeMode when a conflict occurs we'll insert an {@link BatchExecExchange} node
- * with this shuffleMode to resolve conflict
+ * with this exchange mode to resolve conflict
*/
public InputPriorityConflictResolver(
List<ExecNode<?>> roots,
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 c2e01de..3ecf718 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
@@ -22,7 +22,7 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.graph.GlobalDataExchangeMode;
import org.apache.flink.table.api.config.ExecutionConfigOptions;
-/** Utility class to load job-wide shuffle mode. */
+/** Utility class to load job-wide exchange mode. */
public class StreamExchangeModeUtils {
static final String ALL_EDGES_BLOCKING_LEGACY = "batch";