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 <[email protected]> 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); + } +}
