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 2024/06/05 08:43:36 UTC
(flink) branch master updated (2e158fe300f -> c90f30f40bd)
This is an automated email from the ASF dual-hosted git repository.
xtsong pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
from 2e158fe300f [FLINK-35200][table] Fix missing clusterInfo in materialized table refresh rest API return value
new 5c04781401f [refactor][test] Move test to StreamingJobGraphGeneratorWithOperatorAttributesTest
new c90f30f40bd [FLINK-35475][runtime] Introduce isInternalSorterSupport to OperatorAttributes
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../api/operators/OperatorAttributes.java | 12 +-
.../api/operators/OperatorAttributesBuilder.java | 8 +-
.../sortpartition/KeyedSortPartitionOperator.java | 5 +-
.../AbstractMultipleInputTransformation.java | 4 +
.../transformations/OneInputTransformation.java | 4 +
.../transformations/TwoInputTransformation.java | 4 +
.../MultiInputTransformationTranslator.java | 3 +-
.../OneInputTransformationTranslator.java | 14 +-
.../TwoInputTransformationTranslator.java | 5 +-
.../api/graph/StreamingJobGraphGeneratorTest.java | 148 ---------
...obGraphGeneratorWithOperatorAttributesTest.java | 368 +++++++++++++++++++++
11 files changed, 408 insertions(+), 167 deletions(-)
create mode 100644 flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGeneratorWithOperatorAttributesTest.java
(flink) 02/02: [FLINK-35475][runtime] Introduce isInternalSorterSupport to OperatorAttributes
Posted by xt...@apache.org.
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 c90f30f40bda0874bfdb16b0ea20f8a556947abd
Author: sxnan <su...@gmail.com>
AuthorDate: Fri May 31 23:50:37 2024 +0800
[FLINK-35475][runtime] Introduce isInternalSorterSupport to OperatorAttributes
This closes #24874
---
.../api/operators/OperatorAttributes.java | 12 +-
.../api/operators/OperatorAttributesBuilder.java | 8 +-
.../sortpartition/KeyedSortPartitionOperator.java | 5 +-
.../AbstractMultipleInputTransformation.java | 4 +
.../transformations/OneInputTransformation.java | 4 +
.../transformations/TwoInputTransformation.java | 4 +
.../MultiInputTransformationTranslator.java | 3 +-
.../OneInputTransformationTranslator.java | 14 +-
.../TwoInputTransformationTranslator.java | 5 +-
...obGraphGeneratorWithOperatorAttributesTest.java | 174 +++++++++++++++++++++
10 files changed, 214 insertions(+), 19 deletions(-)
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/OperatorAttributes.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/OperatorAttributes.java
index 6c699e15f9f..db883c03631 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/OperatorAttributes.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/OperatorAttributes.java
@@ -30,9 +30,11 @@ public class OperatorAttributes implements Serializable {
private static final long serialVersionUID = 1L;
private final boolean outputOnlyAfterEndOfStream;
+ private final boolean internalSorterSupported;
- OperatorAttributes(boolean outputOnlyAfterEndOfStream) {
+ OperatorAttributes(boolean outputOnlyAfterEndOfStream, boolean internalSorterSupported) {
this.outputOnlyAfterEndOfStream = outputOnlyAfterEndOfStream;
+ this.internalSorterSupported = internalSorterSupported;
}
/**
@@ -49,4 +51,12 @@ public class OperatorAttributes implements Serializable {
public boolean isOutputOnlyAfterEndOfStream() {
return outputOnlyAfterEndOfStream;
}
+
+ /**
+ * Returns true iff the operator uses an internal sorter to sort inputs by key. When it is true,
+ * the input records will not to be sorted externally before being fed into this operator.
+ */
+ public boolean isInternalSorterSupported() {
+ return internalSorterSupported;
+ }
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/OperatorAttributesBuilder.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/OperatorAttributesBuilder.java
index 3239444477f..a504aaa49cd 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/OperatorAttributesBuilder.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/OperatorAttributesBuilder.java
@@ -23,6 +23,7 @@ import org.apache.flink.annotation.Experimental;
@Experimental
public class OperatorAttributesBuilder {
private boolean outputOnlyAfterEndOfStream = false;
+ private boolean internalSorterSupported = false;
/**
* Set to true if and only if the operator only emits records after all its inputs have ended.
@@ -34,7 +35,12 @@ public class OperatorAttributesBuilder {
return this;
}
+ public OperatorAttributesBuilder setInternalSorterSupported(boolean internalSorterSupported) {
+ this.internalSorterSupported = internalSorterSupported;
+ return this;
+ }
+
public OperatorAttributes build() {
- return new OperatorAttributes(outputOnlyAfterEndOfStream);
+ return new OperatorAttributes(outputOnlyAfterEndOfStream, internalSorterSupported);
}
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/sortpartition/KeyedSortPartitionOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/sortpartition/KeyedSortPartitionOperator.java
index 8c45a54ed5a..dc4458d7176 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/sortpartition/KeyedSortPartitionOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/sortpartition/KeyedSortPartitionOperator.java
@@ -250,7 +250,10 @@ public class KeyedSortPartitionOperator<INPUT, KEY> extends AbstractStreamOperat
@Override
public OperatorAttributes getOperatorAttributes() {
- return new OperatorAttributesBuilder().setOutputOnlyAfterEndOfStream(true).build();
+ return new OperatorAttributesBuilder()
+ .setOutputOnlyAfterEndOfStream(true)
+ .setInternalSorterSupported(true)
+ .build();
}
/**
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/AbstractMultipleInputTransformation.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/AbstractMultipleInputTransformation.java
index 154ba63b0f3..7b97bc8ecbd 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/AbstractMultipleInputTransformation.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/AbstractMultipleInputTransformation.java
@@ -94,4 +94,8 @@ public abstract class AbstractMultipleInputTransformation<OUT> extends PhysicalT
public boolean isOutputOnlyAfterEndOfStream() {
return operatorFactory.getOperatorAttributes().isOutputOnlyAfterEndOfStream();
}
+
+ public boolean isInternalSorterSupported() {
+ return operatorFactory.getOperatorAttributes().isInternalSorterSupported();
+ }
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/OneInputTransformation.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/OneInputTransformation.java
index ba8162d8e5b..6aeaf2d96ee 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/OneInputTransformation.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/OneInputTransformation.java
@@ -188,4 +188,8 @@ public class OneInputTransformation<IN, OUT> extends PhysicalTransformation<OUT>
public boolean isOutputOnlyAfterEndOfStream() {
return operatorFactory.getOperatorAttributes().isOutputOnlyAfterEndOfStream();
}
+
+ public boolean isInternalSorterSupported() {
+ return operatorFactory.getOperatorAttributes().isInternalSorterSupported();
+ }
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/TwoInputTransformation.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/TwoInputTransformation.java
index f7b60c15d2f..007f908c809 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/TwoInputTransformation.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/TwoInputTransformation.java
@@ -235,4 +235,8 @@ public class TwoInputTransformation<IN1, IN2, OUT> extends PhysicalTransformatio
public boolean isOutputOnlyAfterEndOfStream() {
return operatorFactory.getOperatorAttributes().isOutputOnlyAfterEndOfStream();
}
+
+ public boolean isInternalSorterSupported() {
+ return operatorFactory.getOperatorAttributes().isInternalSorterSupported();
+ }
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/translators/MultiInputTransformationTranslator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/translators/MultiInputTransformationTranslator.java
index b2a39d1acf4..be2980f3ab6 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/translators/MultiInputTransformationTranslator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/translators/MultiInputTransformationTranslator.java
@@ -132,7 +132,8 @@ public class MultiInputTransformationTranslator<OUT>
private void maybeApplyBatchExecutionSettings(
final AbstractMultipleInputTransformation<OUT> transformation, final Context context) {
- if (transformation instanceof KeyedMultipleInputTransformation) {
+ if (transformation instanceof KeyedMultipleInputTransformation
+ && !transformation.isInternalSorterSupported()) {
KeyedMultipleInputTransformation<OUT> keyedTransformation =
(KeyedMultipleInputTransformation<OUT>) transformation;
List<Transformation<?>> inputs = transformation.getInputs();
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/translators/OneInputTransformationTranslator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/translators/OneInputTransformationTranslator.java
index 804d1f7c09a..6d7ae8103f1 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/translators/OneInputTransformationTranslator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/translators/OneInputTransformationTranslator.java
@@ -22,9 +22,6 @@ import org.apache.flink.annotation.Internal;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.graph.StreamConfig;
import org.apache.flink.streaming.api.graph.TransformationTranslator;
-import org.apache.flink.streaming.api.operators.SimpleOperatorFactory;
-import org.apache.flink.streaming.api.operators.StreamOperatorFactory;
-import org.apache.flink.streaming.api.operators.sortpartition.KeyedSortPartitionOperator;
import org.apache.flink.streaming.api.transformations.OneInputTransformation;
import java.util.Collection;
@@ -80,16 +77,7 @@ public final class OneInputTransformationTranslator<IN, OUT>
private void maybeApplyBatchExecutionSettings(
final OneInputTransformation<IN, OUT> transformation, final Context context) {
KeySelector<IN, ?> keySelector = transformation.getStateKeySelector();
- if (keySelector != null) {
- // KeyedSortPartitionOperator doesn't need sorted input because it sorts data
- // internally.
- StreamOperatorFactory<OUT> operatorFactory = transformation.getOperatorFactory();
- if (operatorFactory instanceof SimpleOperatorFactory
- && operatorFactory.getStreamOperatorClass(
- Thread.currentThread().getContextClassLoader())
- == KeyedSortPartitionOperator.class) {
- return;
- }
+ if (keySelector != null && !transformation.isInternalSorterSupported()) {
BatchExecutionUtils.applyBatchExecutionSettings(
transformation.getId(), context, StreamConfig.InputRequirement.SORTED);
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/translators/TwoInputTransformationTranslator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/translators/TwoInputTransformationTranslator.java
index 13dca02c4e1..8313985e001 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/translators/TwoInputTransformationTranslator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/translators/TwoInputTransformationTranslator.java
@@ -89,8 +89,9 @@ public class TwoInputTransformationTranslator<IN1, IN2, OUT>
? StreamConfig.InputRequirement.SORTED
: StreamConfig.InputRequirement.PASS_THROUGH;
- if (input1Requirement == StreamConfig.InputRequirement.SORTED
- || input2Requirement == StreamConfig.InputRequirement.SORTED) {
+ if ((input1Requirement == StreamConfig.InputRequirement.SORTED
+ || input2Requirement == StreamConfig.InputRequirement.SORTED)
+ && !transformation.isInternalSorterSupported()) {
BatchExecutionUtils.applyBatchExecutionSettings(
transformation.getId(), context, input1Requirement, input2Requirement);
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGeneratorWithOperatorAttributesTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGeneratorWithOperatorAttributesTest.java
index 77536de7bc0..94c9e2ec173 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGeneratorWithOperatorAttributesTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGeneratorWithOperatorAttributesTest.java
@@ -19,6 +19,7 @@
package org.apache.flink.streaming.api.graph;
import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.io.network.partition.ResultPartitionType;
@@ -28,10 +29,15 @@ import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink;
+import org.apache.flink.streaming.api.operators.ChainingStrategy;
import org.apache.flink.streaming.api.operators.OperatorAttributes;
import org.apache.flink.streaming.api.operators.OperatorAttributesBuilder;
import org.apache.flink.streaming.api.operators.StreamMap;
+import org.apache.flink.streaming.api.operators.StreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
import org.apache.flink.streaming.api.operators.co.CoProcessOperator;
+import org.apache.flink.streaming.api.transformations.KeyedMultipleInputTransformation;
import org.apache.flink.util.Collector;
import org.junit.jupiter.api.Test;
@@ -144,6 +150,139 @@ public class StreamingJobGraphGeneratorWithOperatorAttributesTest {
assertThat(node.getManagedMemoryOperatorScopeUseCaseWeights()).hasSize(weightSize);
}
+ @Test
+ void testOneInputOperatorWithInternalSorterSupported() {
+ final StreamExecutionEnvironment env =
+ StreamExecutionEnvironment.getExecutionEnvironment(new Configuration());
+
+ final DataStream<Integer> source1 = env.fromData(1, 2, 3).name("source1");
+ source1.keyBy(x -> x)
+ .transform(
+ "internalSorter",
+ Types.INT,
+ new StreamOperatorWithConfigurableOperatorAttributes<>(
+ (MapFunction<Integer, Integer>) value -> value,
+ new OperatorAttributesBuilder()
+ .setOutputOnlyAfterEndOfStream(true)
+ .setInternalSorterSupported(true)
+ .build()))
+ .keyBy(x -> x)
+ .transform(
+ "noInternalSorter",
+ Types.INT,
+ new StreamOperatorWithConfigurableOperatorAttributes<>(
+ (MapFunction<Integer, Integer>) value -> value,
+ new OperatorAttributesBuilder()
+ .setOutputOnlyAfterEndOfStream(true)
+ .build()))
+ .sinkTo(new DiscardingSink<>())
+ .name("sink");
+
+ final StreamGraph streamGraph = env.getStreamGraph(false);
+ Map<String, StreamNode> nodeMap = new HashMap<>();
+ for (StreamNode node : streamGraph.getStreamNodes()) {
+ nodeMap.put(node.getOperatorName(), node);
+ }
+
+ assertThat(nodeMap.get("internalSorter").getInputRequirements()).isEmpty();
+ assertThat(nodeMap.get("noInternalSorter").getInputRequirements().get(0))
+ .isEqualTo(StreamConfig.InputRequirement.SORTED);
+ }
+
+ @Test
+ void testTwoInputOperatorWithInternalSorterSupported() {
+ final StreamExecutionEnvironment env =
+ StreamExecutionEnvironment.getExecutionEnvironment(new Configuration());
+
+ final DataStream<Integer> source1 = env.fromData(1, 2, 3).name("source1");
+ final DataStream<Integer> source2 = env.fromData(1, 2, 3).name("source2");
+ source1.keyBy(x -> x)
+ .connect(source2.keyBy(x -> x))
+ .transform(
+ "internalSorter",
+ Types.INT,
+ new TwoInputStreamOperatorWithConfigurableOperatorAttributes<>(
+ new OperatorAttributesBuilder()
+ .setOutputOnlyAfterEndOfStream(true)
+ .setInternalSorterSupported(true)
+ .build()))
+ .keyBy(x -> x)
+ .connect(source2.keyBy(x -> x))
+ .transform(
+ "noInternalSorter",
+ Types.INT,
+ new TwoInputStreamOperatorWithConfigurableOperatorAttributes<>(
+ new OperatorAttributesBuilder()
+ .setOutputOnlyAfterEndOfStream(true)
+ .build()))
+ .sinkTo(new DiscardingSink<>())
+ .name("sink");
+
+ final StreamGraph streamGraph = env.getStreamGraph(false);
+ Map<String, StreamNode> nodeMap = new HashMap<>();
+ for (StreamNode node : streamGraph.getStreamNodes()) {
+ nodeMap.put(node.getOperatorName(), node);
+ }
+
+ assertThat(nodeMap.get("internalSorter").getInputRequirements()).isEmpty();
+ assertThat(nodeMap.get("noInternalSorter").getInputRequirements().get(0))
+ .isEqualTo(StreamConfig.InputRequirement.SORTED);
+ assertThat(nodeMap.get("noInternalSorter").getInputRequirements().get(1))
+ .isEqualTo(StreamConfig.InputRequirement.SORTED);
+ }
+
+ @Test
+ void testMultipleInputOperatorWithInternalSorterSupported() {
+ final StreamExecutionEnvironment env =
+ StreamExecutionEnvironment.getExecutionEnvironment(new Configuration());
+
+ final DataStream<Integer> source1 = env.fromData(1, 2, 3).name("source1");
+ final DataStream<Integer> source2 = env.fromData(1, 2, 3).name("source2");
+
+ KeyedMultipleInputTransformation<Integer> transform =
+ new KeyedMultipleInputTransformation<>(
+ "internalSorter",
+ new OperatorAttributesConfigurableOperatorFactory<>(
+ new OperatorAttributesBuilder()
+ .setOutputOnlyAfterEndOfStream(true)
+ .setInternalSorterSupported(true)
+ .build()),
+ BasicTypeInfo.INT_TYPE_INFO,
+ 3,
+ BasicTypeInfo.INT_TYPE_INFO);
+
+ transform.addInput(source1.keyBy(x -> x).getTransformation(), x -> x);
+ transform.addInput(source2.getTransformation(), null);
+
+ KeyedMultipleInputTransformation<Integer> transform2 =
+ new KeyedMultipleInputTransformation<>(
+ "noInternalSorter",
+ new OperatorAttributesConfigurableOperatorFactory<>(
+ new OperatorAttributesBuilder()
+ .setOutputOnlyAfterEndOfStream(true)
+ .build()),
+ BasicTypeInfo.INT_TYPE_INFO,
+ 3,
+ BasicTypeInfo.INT_TYPE_INFO);
+
+ transform2.addInput(transform, null);
+ transform2.addInput(source2.keyBy(x -> x).getTransformation(), x -> x);
+
+ new DataStream<>(env, transform2).sinkTo(new DiscardingSink<>());
+
+ final StreamGraph streamGraph = env.getStreamGraph(false);
+ Map<String, StreamNode> nodeMap = new HashMap<>();
+ for (StreamNode node : streamGraph.getStreamNodes()) {
+ nodeMap.put(node.getOperatorName(), node);
+ }
+
+ assertThat(nodeMap.get("internalSorter").getInputRequirements()).isEmpty();
+ assertThat(nodeMap.get("noInternalSorter").getInputRequirements().get(0))
+ .isEqualTo(StreamConfig.InputRequirement.PASS_THROUGH);
+ assertThat(nodeMap.get("noInternalSorter").getInputRequirements().get(1))
+ .isEqualTo(StreamConfig.InputRequirement.SORTED);
+ }
+
private static class StreamOperatorWithConfigurableOperatorAttributes<IN, OUT>
extends StreamMap<IN, OUT> {
private final OperatorAttributes attributes;
@@ -176,6 +315,41 @@ public class StreamingJobGraphGeneratorWithOperatorAttributesTest {
}
}
+ private static class OperatorAttributesConfigurableOperatorFactory<OUT>
+ implements StreamOperatorFactory<OUT> {
+
+ private final OperatorAttributes operatorAttributes;
+
+ public OperatorAttributesConfigurableOperatorFactory(
+ OperatorAttributes operatorAttributes) {
+ this.operatorAttributes = operatorAttributes;
+ }
+
+ @Override
+ public <T extends StreamOperator<OUT>> T createStreamOperator(
+ StreamOperatorParameters<OUT> parameters) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void setChainingStrategy(ChainingStrategy strategy) {}
+
+ @Override
+ public ChainingStrategy getChainingStrategy() {
+ return ChainingStrategy.DEFAULT_CHAINING_STRATEGY;
+ }
+
+ @Override
+ public Class<? extends StreamOperator> getStreamOperatorClass(ClassLoader classLoader) {
+ return StreamMap.class;
+ }
+
+ @Override
+ public OperatorAttributes getOperatorAttributes() {
+ return operatorAttributes;
+ }
+ }
+
private static class NoOpCoProcessFunction<IN1, IN2, OUT>
extends CoProcessFunction<IN1, IN2, OUT> {
@Override
(flink) 01/02: [refactor][test] Move test to StreamingJobGraphGeneratorWithOperatorAttributesTest
Posted by xt...@apache.org.
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 5c04781401f3894f115051b61214450ea6988fa6
Author: sxnan <su...@gmail.com>
AuthorDate: Fri May 31 23:48:35 2024 +0800
[refactor][test] Move test to StreamingJobGraphGeneratorWithOperatorAttributesTest
---
.../api/graph/StreamingJobGraphGeneratorTest.java | 148 ----------------
...obGraphGeneratorWithOperatorAttributesTest.java | 194 +++++++++++++++++++++
2 files changed, 194 insertions(+), 148 deletions(-)
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 4704cbea5c5..00e5f2aef13 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
@@ -91,7 +91,6 @@ import org.apache.flink.streaming.api.datastream.IterativeStream;
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.CoProcessFunction;
import org.apache.flink.streaming.api.functions.sink.PrintSink;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink;
@@ -104,8 +103,6 @@ import org.apache.flink.streaming.api.operators.ChainingStrategy;
import org.apache.flink.streaming.api.operators.CoordinatedOperatorFactory;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.api.operators.OneInputStreamOperatorFactory;
-import org.apache.flink.streaming.api.operators.OperatorAttributes;
-import org.apache.flink.streaming.api.operators.OperatorAttributesBuilder;
import org.apache.flink.streaming.api.operators.SimpleOperatorFactory;
import org.apache.flink.streaming.api.operators.SourceOperatorFactory;
import org.apache.flink.streaming.api.operators.StreamMap;
@@ -113,7 +110,6 @@ import org.apache.flink.streaming.api.operators.StreamOperator;
import org.apache.flink.streaming.api.operators.StreamOperatorFactory;
import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
import org.apache.flink.streaming.api.operators.YieldingOperatorFactory;
-import org.apache.flink.streaming.api.operators.co.CoProcessOperator;
import org.apache.flink.streaming.api.transformations.CacheTransformation;
import org.apache.flink.streaming.api.transformations.MultipleInputTransformation;
import org.apache.flink.streaming.api.transformations.OneInputTransformation;
@@ -2304,107 +2300,6 @@ class StreamingJobGraphGeneratorTest {
new TestingOutputFormatSupportConcurrentExecutionAttempts<>(), true);
}
- @Test
- void testOutputOnlyAfterEndOfStream() {
- final StreamExecutionEnvironment env =
- StreamExecutionEnvironment.getExecutionEnvironment(new Configuration());
-
- final DataStream<Integer> source = env.fromData(1, 2, 3).name("source");
- source.keyBy(x -> x)
- .transform(
- "transform",
- Types.INT,
- new StreamOperatorWithConfigurableOperatorAttributes<>(
- x -> x,
- new OperatorAttributesBuilder()
- .setOutputOnlyAfterEndOfStream(true)
- .build()))
- .map(x -> x)
- .sinkTo(new DiscardingSink<>())
- .disableChaining()
- .name("sink");
-
- final StreamGraph streamGraph = env.getStreamGraph(false);
- Map<String, StreamNode> nodeMap = new HashMap<>();
- for (StreamNode node : streamGraph.getStreamNodes()) {
- nodeMap.put(node.getOperatorName(), node);
- }
- assertThat(nodeMap).hasSize(4);
- assertThat(nodeMap.get("Source: source").isOutputOnlyAfterEndOfStream()).isFalse();
- assertThat(nodeMap.get("transform").isOutputOnlyAfterEndOfStream()).isTrue();
- assertThat(nodeMap.get("Map").isOutputOnlyAfterEndOfStream()).isFalse();
- assertThat(nodeMap.get("sink: Writer").isOutputOnlyAfterEndOfStream()).isFalse();
- assertManagedMemoryWeightsSize(nodeMap.get("Source: source"), 0);
- assertManagedMemoryWeightsSize(nodeMap.get("transform"), 1);
- assertManagedMemoryWeightsSize(nodeMap.get("Map"), 0);
- assertManagedMemoryWeightsSize(nodeMap.get("sink: Writer"), 0);
-
- JobGraph jobGraph = StreamingJobGraphGenerator.createJobGraph(streamGraph);
- Map<String, JobVertex> vertexMap = new HashMap<>();
- for (JobVertex vertex : jobGraph.getVertices()) {
- vertexMap.put(vertex.getName(), vertex);
- }
- assertThat(vertexMap).hasSize(3);
- assertHasOutputPartitionType(
- vertexMap.get("Source: source"), ResultPartitionType.PIPELINED_BOUNDED);
- assertHasOutputPartitionType(
- vertexMap.get("transform -> Map"), ResultPartitionType.BLOCKING);
- assertThat(vertexMap.get("Source: source").isAnyOutputBlocking()).isFalse();
- assertThat(vertexMap.get("transform -> Map").isAnyOutputBlocking()).isTrue();
- assertThat(vertexMap.get("sink: Writer").isAnyOutputBlocking()).isFalse();
-
- env.disableOperatorChaining();
- jobGraph = StreamingJobGraphGenerator.createJobGraph(env.getStreamGraph(false));
- vertexMap = new HashMap<>();
- for (JobVertex vertex : jobGraph.getVertices()) {
- vertexMap.put(vertex.getName(), vertex);
- }
- assertThat(vertexMap).hasSize(4);
- assertHasOutputPartitionType(
- vertexMap.get("Source: source"), ResultPartitionType.PIPELINED_BOUNDED);
- assertHasOutputPartitionType(vertexMap.get("transform"), ResultPartitionType.BLOCKING);
- assertHasOutputPartitionType(vertexMap.get("Map"), ResultPartitionType.PIPELINED_BOUNDED);
- assertThat(vertexMap.get("Source: source").isAnyOutputBlocking()).isFalse();
- assertThat(vertexMap.get("transform").isAnyOutputBlocking()).isTrue();
- assertThat(vertexMap.get("Map").isAnyOutputBlocking()).isFalse();
- assertThat(vertexMap.get("sink: Writer").isAnyOutputBlocking()).isFalse();
- }
-
- @Test
- void testApplyBatchExecutionSettingsOnTwoInputOperator() {
- final StreamExecutionEnvironment env =
- StreamExecutionEnvironment.getExecutionEnvironment(new Configuration());
-
- final DataStream<Integer> source1 = env.fromData(1, 2, 3).name("source1");
- final DataStream<Integer> source2 = env.fromData(1, 2, 3).name("source2");
- source1.keyBy(x -> x)
- .connect(source2.keyBy(x -> x))
- .transform(
- "transform",
- Types.INT,
- new TwoInputStreamOperatorWithConfigurableOperatorAttributes<>(
- new OperatorAttributesBuilder()
- .setOutputOnlyAfterEndOfStream(true)
- .build()))
- .sinkTo(new DiscardingSink<>())
- .name("sink");
-
- final StreamGraph streamGraph = env.getStreamGraph(false);
- Map<String, StreamNode> nodeMap = new HashMap<>();
- for (StreamNode node : streamGraph.getStreamNodes()) {
- nodeMap.put(node.getOperatorName(), node);
- }
- assertThat(nodeMap).hasSize(4);
- assertManagedMemoryWeightsSize(nodeMap.get("Source: source1"), 0);
- assertManagedMemoryWeightsSize(nodeMap.get("Source: source2"), 0);
- assertManagedMemoryWeightsSize(nodeMap.get("transform"), 1);
- assertManagedMemoryWeightsSize(nodeMap.get("sink: Writer"), 0);
- }
-
- private void assertManagedMemoryWeightsSize(StreamNode node, int weightSize) {
- assertThat(node.getManagedMemoryOperatorScopeUseCaseWeights()).hasSize(weightSize);
- }
-
private static void testWhetherOutputFormatSupportsConcurrentExecutionAttempts(
OutputFormat<Integer> outputFormat, boolean isSupported) {
final StreamExecutionEnvironment env =
@@ -3047,47 +2942,4 @@ class StreamingJobGraphGeneratorTest {
@Override
public void cancel() {}
}
-
- private static class StreamOperatorWithConfigurableOperatorAttributes<IN, OUT>
- extends StreamMap<IN, OUT> {
- private final OperatorAttributes attributes;
-
- public StreamOperatorWithConfigurableOperatorAttributes(
- MapFunction<IN, OUT> mapper, OperatorAttributes attributes) {
- super(mapper);
- this.attributes = attributes;
- }
-
- @Override
- public OperatorAttributes getOperatorAttributes() {
- return attributes;
- }
- }
-
- private static class TwoInputStreamOperatorWithConfigurableOperatorAttributes<IN1, IN2, OUT>
- extends CoProcessOperator<IN1, IN2, OUT> {
- private final OperatorAttributes attributes;
-
- public TwoInputStreamOperatorWithConfigurableOperatorAttributes(
- OperatorAttributes attributes) {
- super(new NoOpCoProcessFunction<>());
- this.attributes = attributes;
- }
-
- @Override
- public OperatorAttributes getOperatorAttributes() {
- return attributes;
- }
- }
-
- private static class NoOpCoProcessFunction<IN1, IN2, OUT>
- extends CoProcessFunction<IN1, IN2, OUT> {
- @Override
- public void processElement1(
- IN1 value, CoProcessFunction<IN1, IN2, OUT>.Context ctx, Collector<OUT> out) {}
-
- @Override
- public void processElement2(
- IN2 value, CoProcessFunction<IN1, IN2, OUT>.Context ctx, Collector<OUT> out) {}
- }
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGeneratorWithOperatorAttributesTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGeneratorWithOperatorAttributesTest.java
new file mode 100644
index 00000000000..77536de7bc0
--- /dev/null
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGeneratorWithOperatorAttributesTest.java
@@ -0,0 +1,194 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.api.graph;
+
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionType;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.jobgraph.JobVertex;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
+import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink;
+import org.apache.flink.streaming.api.operators.OperatorAttributes;
+import org.apache.flink.streaming.api.operators.OperatorAttributesBuilder;
+import org.apache.flink.streaming.api.operators.StreamMap;
+import org.apache.flink.streaming.api.operators.co.CoProcessOperator;
+import org.apache.flink.util.Collector;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link StreamingJobGraphGenerator} with internal sorter. */
+public class StreamingJobGraphGeneratorWithOperatorAttributesTest {
+ @Test
+ void testOutputOnlyAfterEndOfStream() {
+ final StreamExecutionEnvironment env =
+ StreamExecutionEnvironment.getExecutionEnvironment(new Configuration());
+
+ final DataStream<Integer> source = env.fromData(1, 2, 3).name("source");
+ source.keyBy(x -> x)
+ .transform(
+ "transform",
+ Types.INT,
+ new StreamOperatorWithConfigurableOperatorAttributes<>(
+ x -> x,
+ new OperatorAttributesBuilder()
+ .setOutputOnlyAfterEndOfStream(true)
+ .build()))
+ .map(x -> x)
+ .sinkTo(new DiscardingSink<>())
+ .disableChaining()
+ .name("sink");
+
+ final StreamGraph streamGraph = env.getStreamGraph(false);
+ Map<String, StreamNode> nodeMap = new HashMap<>();
+ for (StreamNode node : streamGraph.getStreamNodes()) {
+ nodeMap.put(node.getOperatorName(), node);
+ }
+ assertThat(nodeMap).hasSize(4);
+ assertThat(nodeMap.get("Source: source").isOutputOnlyAfterEndOfStream()).isFalse();
+ assertThat(nodeMap.get("transform").isOutputOnlyAfterEndOfStream()).isTrue();
+ assertThat(nodeMap.get("Map").isOutputOnlyAfterEndOfStream()).isFalse();
+ assertThat(nodeMap.get("sink: Writer").isOutputOnlyAfterEndOfStream()).isFalse();
+ assertManagedMemoryWeightsSize(nodeMap.get("Source: source"), 0);
+ assertManagedMemoryWeightsSize(nodeMap.get("transform"), 1);
+ assertManagedMemoryWeightsSize(nodeMap.get("Map"), 0);
+ assertManagedMemoryWeightsSize(nodeMap.get("sink: Writer"), 0);
+
+ JobGraph jobGraph = StreamingJobGraphGenerator.createJobGraph(streamGraph);
+ Map<String, JobVertex> vertexMap = new HashMap<>();
+ for (JobVertex vertex : jobGraph.getVertices()) {
+ vertexMap.put(vertex.getName(), vertex);
+ }
+ assertThat(vertexMap).hasSize(3);
+ assertHasOutputPartitionType(
+ vertexMap.get("Source: source"), ResultPartitionType.PIPELINED_BOUNDED);
+ assertHasOutputPartitionType(
+ vertexMap.get("transform -> Map"), ResultPartitionType.BLOCKING);
+ assertThat(vertexMap.get("Source: source").isAnyOutputBlocking()).isFalse();
+ assertThat(vertexMap.get("transform -> Map").isAnyOutputBlocking()).isTrue();
+ assertThat(vertexMap.get("sink: Writer").isAnyOutputBlocking()).isFalse();
+
+ env.disableOperatorChaining();
+ jobGraph = StreamingJobGraphGenerator.createJobGraph(env.getStreamGraph(false));
+ vertexMap = new HashMap<>();
+ for (JobVertex vertex : jobGraph.getVertices()) {
+ vertexMap.put(vertex.getName(), vertex);
+ }
+ assertThat(vertexMap).hasSize(4);
+ assertHasOutputPartitionType(
+ vertexMap.get("Source: source"), ResultPartitionType.PIPELINED_BOUNDED);
+ assertHasOutputPartitionType(vertexMap.get("transform"), ResultPartitionType.BLOCKING);
+ assertHasOutputPartitionType(vertexMap.get("Map"), ResultPartitionType.PIPELINED_BOUNDED);
+ assertThat(vertexMap.get("Source: source").isAnyOutputBlocking()).isFalse();
+ assertThat(vertexMap.get("transform").isAnyOutputBlocking()).isTrue();
+ assertThat(vertexMap.get("Map").isAnyOutputBlocking()).isFalse();
+ assertThat(vertexMap.get("sink: Writer").isAnyOutputBlocking()).isFalse();
+ }
+
+ @Test
+ void testApplyBatchExecutionSettingsOnTwoInputOperator() {
+ final StreamExecutionEnvironment env =
+ StreamExecutionEnvironment.getExecutionEnvironment(new Configuration());
+
+ final DataStream<Integer> source1 = env.fromData(1, 2, 3).name("source1");
+ final DataStream<Integer> source2 = env.fromData(1, 2, 3).name("source2");
+ source1.keyBy(x -> x)
+ .connect(source2.keyBy(x -> x))
+ .transform(
+ "transform",
+ Types.INT,
+ new TwoInputStreamOperatorWithConfigurableOperatorAttributes<>(
+ new OperatorAttributesBuilder()
+ .setOutputOnlyAfterEndOfStream(true)
+ .build()))
+ .sinkTo(new DiscardingSink<>())
+ .name("sink");
+
+ final StreamGraph streamGraph = env.getStreamGraph(false);
+ Map<String, StreamNode> nodeMap = new HashMap<>();
+ for (StreamNode node : streamGraph.getStreamNodes()) {
+ nodeMap.put(node.getOperatorName(), node);
+ }
+ assertThat(nodeMap).hasSize(4);
+ assertManagedMemoryWeightsSize(nodeMap.get("Source: source1"), 0);
+ assertManagedMemoryWeightsSize(nodeMap.get("Source: source2"), 0);
+ assertManagedMemoryWeightsSize(nodeMap.get("transform"), 1);
+ assertManagedMemoryWeightsSize(nodeMap.get("sink: Writer"), 0);
+ }
+
+ private static void assertManagedMemoryWeightsSize(StreamNode node, int weightSize) {
+ assertThat(node.getManagedMemoryOperatorScopeUseCaseWeights()).hasSize(weightSize);
+ }
+
+ private static class StreamOperatorWithConfigurableOperatorAttributes<IN, OUT>
+ extends StreamMap<IN, OUT> {
+ private final OperatorAttributes attributes;
+
+ public StreamOperatorWithConfigurableOperatorAttributes(
+ MapFunction<IN, OUT> mapper, OperatorAttributes attributes) {
+ super(mapper);
+ this.attributes = attributes;
+ }
+
+ @Override
+ public OperatorAttributes getOperatorAttributes() {
+ return attributes;
+ }
+ }
+
+ private static class TwoInputStreamOperatorWithConfigurableOperatorAttributes<IN1, IN2, OUT>
+ extends CoProcessOperator<IN1, IN2, OUT> {
+ private final OperatorAttributes attributes;
+
+ public TwoInputStreamOperatorWithConfigurableOperatorAttributes(
+ OperatorAttributes attributes) {
+ super(new NoOpCoProcessFunction<>());
+ this.attributes = attributes;
+ }
+
+ @Override
+ public OperatorAttributes getOperatorAttributes() {
+ return attributes;
+ }
+ }
+
+ private static class NoOpCoProcessFunction<IN1, IN2, OUT>
+ extends CoProcessFunction<IN1, IN2, OUT> {
+ @Override
+ public void processElement1(
+ IN1 value, CoProcessFunction<IN1, IN2, OUT>.Context ctx, Collector<OUT> out) {}
+
+ @Override
+ public void processElement2(
+ IN2 value, CoProcessFunction<IN1, IN2, OUT>.Context ctx, Collector<OUT> out) {}
+ }
+
+ private void assertHasOutputPartitionType(
+ JobVertex jobVertex, ResultPartitionType partitionType) {
+ assertThat(jobVertex.getProducedDataSets().get(0).getResultType()).isEqualTo(partitionType);
+ }
+}