This is an automated email from the ASF dual-hosted git repository. arvid pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit d7bfa77bcabb9b52b7d2420efdeda72dea420506 Author: Arvid Heise <[email protected]> AuthorDate: Mon Nov 25 14:39:48 2024 +0100 [FLINK-36788] Fix GlobalCommitter expansion So far, the global committer didn't properly get all properties set during expansion. In some cases, it didn't get expanded at all. This commit also inline SinkTransformationTranslatorITCaseBase.java after the respective V1 implementation was removed. V2 now has the same coverage as V1 used to have. --- .../GlobalCommitterTransformationTranslator.java | 22 +- .../translators/SinkTransformationTranslator.java | 16 ++ .../SinkTransformationTranslatorITCaseBase.java | 256 --------------------- .../SinkV2TransformationTranslatorITCase.java | 253 ++++++++++++++++++-- .../runtime/operators/sink/TestSinkV2.java | 4 +- 5 files changed, 274 insertions(+), 277 deletions(-) diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/translators/GlobalCommitterTransformationTranslator.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/translators/GlobalCommitterTransformationTranslator.java index e4b3449337d..278a2947789 100644 --- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/translators/GlobalCommitterTransformationTranslator.java +++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/translators/GlobalCommitterTransformationTranslator.java @@ -34,11 +34,14 @@ import org.apache.flink.streaming.runtime.operators.sink.GlobalCommitterOperator import org.apache.flink.streaming.runtime.operators.sink.SinkWriterOperatorFactory; import java.util.ArrayDeque; +import java.util.Arrays; import java.util.Collection; import java.util.Collections; import java.util.HashSet; import java.util.Queue; import java.util.Set; +import java.util.function.Consumer; +import java.util.function.Supplier; import static org.apache.flink.streaming.api.connector.sink2.StandardSinkTopologies.GLOBAL_COMMITTER_TRANSFORMATION_NAME; @@ -74,11 +77,10 @@ public class GlobalCommitterTransformationTranslator<CommT> boolean commitOnInput = batch || !checkpointingEnabled || hasUpstreamCommitter(inputStream); // Create a global shuffle and add the global committer with parallelism 1. + DataStream<CommittableMessage<CommT>> global = inputStream.global(); final PhysicalTransformation<Void> transformation = (PhysicalTransformation<Void>) - inputStream - .global() - .transform( + global.transform( GLOBAL_COMMITTER_TRANSFORMATION_NAME, Types.VOID, new GlobalCommitterOperator<>( @@ -87,10 +89,20 @@ public class GlobalCommitterTransformationTranslator<CommT> commitOnInput)) .getTransformation(); transformation.setChainingStrategy(ChainingStrategy.ALWAYS); - transformation.setName(GLOBAL_COMMITTER_TRANSFORMATION_NAME); transformation.setParallelism(1); transformation.setMaxParallelism(1); - return Collections.emptyList(); + copySafely(transformation::setName, globalCommitterTransform::getName); + copySafely(transformation::setUid, globalCommitterTransform::getUid); + copySafely(transformation::setUidHash, globalCommitterTransform::getUserProvidedNodeHash); + + return Arrays.asList(global.getId(), transformation.getId()); + } + + private static <T> void copySafely(Consumer<T> consumer, Supplier<T> provider) { + T value = provider.get(); + if (value != null) { + consumer.accept(value); + } } /** diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/translators/SinkTransformationTranslator.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/translators/SinkTransformationTranslator.java index 52449be1f18..dd41218a497 100644 --- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/translators/SinkTransformationTranslator.java +++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/translators/SinkTransformationTranslator.java @@ -59,6 +59,8 @@ import java.util.Queue; import java.util.Set; import java.util.function.BiConsumer; import java.util.function.Function; +import java.util.function.Supplier; +import java.util.stream.Collectors; import static org.apache.flink.util.Preconditions.checkState; @@ -177,6 +179,12 @@ public class SinkTransformationTranslator<Input, Output> getSinkTransformations(sizeBefore).forEach(context::transform); + repeatUntilConverged( + () -> + getSinkTransformations(sizeBefore).stream() + .flatMap(t -> context.transform(t).stream()) + .collect(Collectors.toList())); + disallowUnalignedCheckpoint(getSinkTransformations(sizeBefore)); // Remove all added sink subtransformations to avoid duplications and allow additional @@ -188,6 +196,14 @@ public class SinkTransformationTranslator<Input, Output> } } + private <R> void repeatUntilConverged(Supplier<R> producer) { + R lastResult = producer.get(); + R nextResult; + while (!lastResult.equals(nextResult = producer.get())) { + lastResult = nextResult; + } + } + private List<Transformation<?>> getSinkTransformations(int sizeBefore) { return executionEnvironment .getTransformations() diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/api/graph/SinkTransformationTranslatorITCaseBase.java b/flink-runtime/src/test/java/org/apache/flink/streaming/api/graph/SinkTransformationTranslatorITCaseBase.java deleted file mode 100644 index e95523a1492..00000000000 --- a/flink-runtime/src/test/java/org/apache/flink/streaming/api/graph/SinkTransformationTranslatorITCaseBase.java +++ /dev/null @@ -1,256 +0,0 @@ -/* - * 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.RuntimeExecutionMode; -import org.apache.flink.api.common.typeutils.base.IntSerializer; -import org.apache.flink.configuration.Configuration; -import org.apache.flink.configuration.ExecutionOptions; -import org.apache.flink.core.io.SimpleVersionedSerializerTypeSerializerProxy; -import org.apache.flink.streaming.api.datastream.DataStream; -import org.apache.flink.streaming.api.datastream.DataStreamSink; -import org.apache.flink.streaming.api.datastream.DataStreamSource; -import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; -import org.apache.flink.streaming.api.operators.ChainingStrategy; -import org.apache.flink.streaming.api.operators.StreamOperatorFactory; -import org.apache.flink.streaming.runtime.operators.sink.CommitterOperatorFactory; -import org.apache.flink.streaming.runtime.operators.sink.SinkWriterOperatorFactory; -import org.apache.flink.testutils.junit.extensions.parameterized.Parameter; -import org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension; -import org.apache.flink.testutils.junit.extensions.parameterized.Parameters; - -import org.junit.jupiter.api.TestTemplate; -import org.junit.jupiter.api.extension.ExtendWith; - -import java.util.Arrays; -import java.util.Collection; -import java.util.function.Predicate; - -import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.Assertions.assertThatThrownBy; - -/** - * Tests for {@link org.apache.flink.streaming.api.transformations.SinkTransformation}. - * - * <p>ATTENTION: This test is extremely brittle. Do NOT remove, add or re-order test cases. - */ -@ExtendWith(ParameterizedTestExtension.class) -abstract class SinkTransformationTranslatorITCaseBase<SinkT> { - - @Parameters(name = "Execution Mode: {0}") - private static Collection<Object> data() { - return Arrays.asList(RuntimeExecutionMode.STREAMING, RuntimeExecutionMode.BATCH); - } - - @Parameter protected RuntimeExecutionMode runtimeExecutionMode; - - static final String NAME = "FileSink"; - static final String SLOT_SHARE_GROUP = "FileGroup"; - static final String UID = "FileUid"; - static final int PARALLELISM = 2; - - abstract SinkT simpleSink(); - - abstract SinkT sinkWithCommitter(); - - abstract DataStreamSink<Integer> sinkTo(DataStream<Integer> stream, SinkT sink); - - @TestTemplate - void generateWriterTopology() { - final StreamGraph streamGraph = buildGraph(simpleSink(), runtimeExecutionMode); - - final StreamNode sourceNode = findNodeName(streamGraph, node -> node.contains("Source")); - final StreamNode writerNode = findWriter(streamGraph); - - assertThat(streamGraph.getStreamNodes()).hasSize(2); - - validateTopology( - sourceNode, - IntSerializer.class, - writerNode, - SinkWriterOperatorFactory.class, - PARALLELISM, - -1); - } - - @TestTemplate - void generateWriterCommitterTopology() { - - final StreamGraph streamGraph = buildGraph(sinkWithCommitter(), runtimeExecutionMode); - - final StreamNode sourceNode = findNodeName(streamGraph, node -> node.contains("Source")); - final StreamNode writerNode = findWriter(streamGraph); - - validateTopology( - sourceNode, - IntSerializer.class, - writerNode, - SinkWriterOperatorFactory.class, - PARALLELISM, - -1); - - final StreamNode committerNode = - findNodeName(streamGraph, name -> name.contains("Committer")); - - assertThat(streamGraph.getStreamNodes()).hasSize(3); - assertNoUnalignedOutput(writerNode); - - validateTopology( - writerNode, - SimpleVersionedSerializerTypeSerializerProxy.class, - committerNode, - CommitterOperatorFactory.class, - PARALLELISM, - -1); - } - - @TestTemplate - void testParallelismConfigured() { - testParallelismConfiguredInternal(true); - - testParallelismConfiguredInternal(false); - } - - private void testParallelismConfiguredInternal(boolean setSinkParallelism) { - final StreamGraph streamGraph = - buildGraph(sinkWithCommitter(), runtimeExecutionMode, setSinkParallelism); - - final StreamNode writerNode = findWriter(streamGraph); - final StreamNode committerNode = findCommitter(streamGraph); - - assertThat(writerNode.isParallelismConfigured()).isEqualTo(setSinkParallelism); - assertThat(committerNode.isParallelismConfigured()).isEqualTo(setSinkParallelism); - } - - StreamNode findWriter(StreamGraph streamGraph) { - return findNodeName( - streamGraph, name -> name.contains("Writer") && !name.contains("Committer")); - } - - StreamNode findCommitter(StreamGraph streamGraph) { - return findNodeName( - streamGraph, - name -> name.contains("Committer") && !name.contains("Global Committer")); - } - - StreamNode findGlobalCommitter(StreamGraph streamGraph) { - return findNodeName(streamGraph, name -> name.contains("Global Committer")); - } - - @TestTemplate - void throwExceptionWithoutSettingUid() { - StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); - - final Configuration config = new Configuration(); - config.set(ExecutionOptions.RUNTIME_MODE, runtimeExecutionMode); - env.configure(config, getClass().getClassLoader()); - // disable auto generating uid - env.getConfig().disableAutoGeneratedUIDs(); - sinkTo(env.fromElements(1, 2), simpleSink()); - assertThatThrownBy(env::getStreamGraph).isInstanceOf(IllegalStateException.class); - } - - @TestTemplate - void disableOperatorChain() { - StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); - - final DataStreamSource<Integer> src = env.fromElements(1, 2); - final DataStreamSink<Integer> dataStreamSink = sinkTo(src, sinkWithCommitter()).name(NAME); - dataStreamSink.disableChaining(); - - final StreamGraph streamGraph = env.getStreamGraph(); - final StreamNode writer = findWriter(streamGraph); - final StreamNode committer = findCommitter(streamGraph); - - assertThat(writer.getOperatorFactory().getChainingStrategy()) - .isEqualTo(ChainingStrategy.NEVER); - assertThat(committer.getOperatorFactory().getChainingStrategy()) - .isEqualTo(ChainingStrategy.NEVER); - } - - void validateTopology( - StreamNode src, - Class<?> srcOutTypeInfo, - StreamNode dest, - Class<? extends StreamOperatorFactory> operatorFactoryClass, - int expectedParallelism, - int expectedMaxParallelism) { - - // verify src node - final StreamEdge srcOutEdge = src.getOutEdges().get(0); - assertThat(srcOutEdge.getTargetId()).isEqualTo(dest.getId()); - assertThat(src.getTypeSerializerOut()).isInstanceOf(srcOutTypeInfo); - - // verify dest node input - final StreamEdge destInputEdge = dest.getInEdges().get(0); - assertThat(destInputEdge.getTargetId()).isEqualTo(dest.getId()); - assertThat(dest.getTypeSerializersIn()[0]).isInstanceOf(srcOutTypeInfo); - - // make sure 2 sink operators have different names/uid - assertThat(dest.getOperatorName()).isNotEqualTo(src.getOperatorName()); - assertThat(dest.getTransformationUID()).isNotEqualTo(src.getTransformationUID()); - - assertThat(dest.getOperatorFactory()).isInstanceOf(operatorFactoryClass); - assertThat(dest.getParallelism()).isEqualTo(expectedParallelism); - assertThat(dest.getMaxParallelism()).isEqualTo(expectedMaxParallelism); - assertThat(dest.getOperatorFactory().getChainingStrategy()) - .isEqualTo(ChainingStrategy.ALWAYS); - assertThat(dest.getSlotSharingGroup()).isEqualTo(SLOT_SHARE_GROUP); - } - - protected static void assertNoUnalignedOutput(StreamNode src) { - assertThat(src.getOutEdges()).allMatch(e -> !e.supportsUnalignedCheckpoints()); - } - - StreamGraph buildGraph(SinkT sink, RuntimeExecutionMode runtimeExecutionMode) { - return buildGraph(sink, runtimeExecutionMode, true); - } - - StreamGraph buildGraph( - SinkT sink, RuntimeExecutionMode runtimeExecutionMode, boolean setSinkParallelism) { - StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); - - final Configuration config = new Configuration(); - config.set(ExecutionOptions.RUNTIME_MODE, runtimeExecutionMode); - env.configure(config, getClass().getClassLoader()); - final DataStreamSource<Integer> src = env.fromElements(1, 2); - final DataStreamSink<Integer> dataStreamSink = sinkTo(src.rebalance(), sink); - setSinkProperty(dataStreamSink, setSinkParallelism); - // Trigger the plan generation but do not clear the transformations - env.getExecutionPlan(); - return env.getStreamGraph(); - } - - private void setSinkProperty( - DataStreamSink<Integer> dataStreamSink, boolean setSinkParallelism) { - dataStreamSink.name(NAME); - dataStreamSink.uid(UID); - if (setSinkParallelism) { - dataStreamSink.setParallelism(SinkTransformationTranslatorITCaseBase.PARALLELISM); - } - dataStreamSink.slotSharingGroup(SLOT_SHARE_GROUP); - } - - StreamNode findNodeName(StreamGraph streamGraph, Predicate<String> predicate) { - return streamGraph.getStreamNodes().stream() - .filter(node -> predicate.test(node.getOperatorName())) - .findFirst() - .orElseThrow(() -> new IllegalStateException("Can not find the node")); - } -} diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/SinkV2TransformationTranslatorITCase.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/SinkV2TransformationTranslatorITCase.java index 6b09b83961d..9e3753031c0 100644 --- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/SinkV2TransformationTranslatorITCase.java +++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/SinkV2TransformationTranslatorITCase.java @@ -18,77 +18,300 @@ package org.apache.flink.streaming.api.graph; +import org.apache.flink.api.common.RuntimeExecutionMode; +import org.apache.flink.api.common.typeutils.base.IntSerializer; import org.apache.flink.api.connector.sink2.Sink; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.ExecutionOptions; +import org.apache.flink.core.io.SimpleVersionedSerializerTypeSerializerProxy; import org.apache.flink.streaming.api.datastream.CustomSinkOperatorUidHashes; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.datastream.DataStreamSink; import org.apache.flink.streaming.api.datastream.DataStreamSource; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.streaming.api.operators.ChainingStrategy; +import org.apache.flink.streaming.api.operators.StreamOperatorFactory; +import org.apache.flink.streaming.runtime.operators.sink.CommitterOperatorFactory; +import org.apache.flink.streaming.runtime.operators.sink.SinkWriterOperatorFactory; import org.apache.flink.streaming.runtime.operators.sink.TestSinkV2; -import org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension; -import org.junit.jupiter.api.TestTemplate; -import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; +import org.junit.jupiter.params.provider.EnumSource; + +import java.util.function.Predicate; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; /** * Tests for {@link org.apache.flink.streaming.api.transformations.SinkTransformation}. * * <p>ATTENTION: This test is extremely brittle. Do NOT remove, add or re-order test cases. */ -@ExtendWith(ParameterizedTestExtension.class) -class SinkV2TransformationTranslatorITCase - extends SinkTransformationTranslatorITCaseBase<Sink<Integer>> { +class SinkV2TransformationTranslatorITCase { + + static final String NAME = "FileSink"; + static final String SLOT_SHARE_GROUP = "FileGroup"; + static final String UID = "FileUid"; + static final int PARALLELISM = 2; + + protected static void assertNoUnalignedOutput(StreamNode src) { + assertThat(src.getOutEdges()).allMatch(e -> !e.supportsUnalignedCheckpoints()); + } - @Override Sink<Integer> simpleSink() { return TestSinkV2.<Integer>newBuilder().build(); } - @Override Sink<Integer> sinkWithCommitter() { return TestSinkV2.<Integer>newBuilder().setDefaultCommitter().build(); } - @Override + Sink<Integer> sinkWithCommitterAndGlobalCommitter() { + return TestSinkV2.<Integer>newBuilder() + .setDefaultCommitter() + .setWithPostCommitTopology(true) + .build(); + } + DataStreamSink<Integer> sinkTo(DataStream<Integer> stream, Sink<Integer> sink) { return stream.sinkTo(sink); } - @TestTemplate + @Test void testSettingOperatorUidHash() { final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); - final DataStreamSource<Integer> src = env.fromElements(1, 2); + final DataStreamSource<Integer> src = env.fromData(1, 2); final String writerHash = "f6b178ce445dc3ffaa06bad27a51fead"; final String committerHash = "68ac8ae79eae4e3135a54f9689c4aa10"; + final String globalCommitterHash = "77e6aa6eeb1643b3765e1e4a7a672f37"; final CustomSinkOperatorUidHashes operatorsUidHashes = CustomSinkOperatorUidHashes.builder() .setWriterUidHash(writerHash) .setCommitterUidHash(committerHash) + .setGlobalCommitterUidHash(globalCommitterHash) .build(); - src.sinkTo(sinkWithCommitter(), operatorsUidHashes).name(NAME); + src.sinkTo(sinkWithCommitterAndGlobalCommitter(), operatorsUidHashes).name(NAME); final StreamGraph streamGraph = env.getStreamGraph(); assertThat(findWriter(streamGraph).getUserHash()).isEqualTo(writerHash); assertThat(findCommitter(streamGraph).getUserHash()).isEqualTo(committerHash); + assertThat(findGlobalCommitter(streamGraph).getUserHash()).isEqualTo(globalCommitterHash); } /** * When ever you need to change something in this test case please think about possible state * upgrade problems introduced by your changes. */ - @TestTemplate + @Test void testSettingOperatorUids() { final String sinkUid = "f6b178ce445dc3ffaa06bad27a51fead"; final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); - final DataStreamSource<Integer> src = env.fromElements(1, 2); - src.sinkTo(sinkWithCommitter()).name(NAME).uid(sinkUid); + final DataStreamSource<Integer> src = env.fromData(1, 2); + src.sinkTo(sinkWithCommitterAndGlobalCommitter()).name(NAME).uid(sinkUid); final StreamGraph streamGraph = env.getStreamGraph(); assertThat(findWriter(streamGraph).getTransformationUID()).isEqualTo(sinkUid); assertThat(findCommitter(streamGraph).getTransformationUID()) .isEqualTo(String.format("Sink Committer: %s", sinkUid)); + assertThat(findGlobalCommitter(streamGraph).getTransformationUID()) + .isEqualTo(String.format("Sink %s Global Committer", sinkUid)); + } + + @Test + void testSettingOperatorNames() { + final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); + final DataStreamSource<Integer> src = env.fromData(1, 2); + src.sinkTo(sinkWithCommitterAndGlobalCommitter()).name(NAME); + + final StreamGraph streamGraph = env.getStreamGraph(); + assertThat(findWriter(streamGraph).getOperatorName()) + .isEqualTo(String.format("%s: Writer", NAME)); + assertThat(findCommitter(streamGraph).getOperatorName()) + .isEqualTo(String.format("%s: Committer", NAME)); + assertThat(findGlobalCommitter(streamGraph).getOperatorName()) + .isEqualTo(String.format("%s: Global Committer", NAME)); + } + + @ParameterizedTest + @EnumSource(RuntimeExecutionMode.class) + void generateWriterTopology(RuntimeExecutionMode runtimeExecutionMode) { + final StreamGraph streamGraph = buildGraph(simpleSink(), runtimeExecutionMode); + + final StreamNode sourceNode = findNodeName(streamGraph, node -> node.contains("Source")); + final StreamNode writerNode = findWriter(streamGraph); + + assertThat(streamGraph.getStreamNodes()).hasSize(2); + + validateTopology( + sourceNode, + IntSerializer.class, + writerNode, + SinkWriterOperatorFactory.class, + PARALLELISM, + -1); + } + + @ParameterizedTest + @EnumSource(RuntimeExecutionMode.class) + void generateWriterCommitterTopology(RuntimeExecutionMode runtimeExecutionMode) { + final StreamGraph streamGraph = buildGraph(sinkWithCommitter(), runtimeExecutionMode); + + final StreamNode sourceNode = findNodeName(streamGraph, node -> node.contains("Source")); + final StreamNode writerNode = findWriter(streamGraph); + + validateTopology( + sourceNode, + IntSerializer.class, + writerNode, + SinkWriterOperatorFactory.class, + PARALLELISM, + -1); + + final StreamNode committerNode = + findNodeName(streamGraph, name -> name.contains("Committer")); + + assertThat(streamGraph.getStreamNodes()).hasSize(3); + assertNoUnalignedOutput(writerNode); + + validateTopology( + writerNode, + SimpleVersionedSerializerTypeSerializerProxy.class, + committerNode, + CommitterOperatorFactory.class, + PARALLELISM, + -1); + } + + @ParameterizedTest + @CsvSource({"STREAMING, true", "STREAMING, false", "BATCH, true", "BATCH, false"}) + void testParallelismConfigured( + RuntimeExecutionMode runtimeExecutionMode, boolean setSinkParallelism) { + final StreamGraph streamGraph = + buildGraph(sinkWithCommitter(), runtimeExecutionMode, setSinkParallelism); + + final StreamNode writerNode = findWriter(streamGraph); + final StreamNode committerNode = findCommitter(streamGraph); + + assertThat(writerNode.isParallelismConfigured()).isEqualTo(setSinkParallelism); + assertThat(committerNode.isParallelismConfigured()).isEqualTo(setSinkParallelism); + } + + @ParameterizedTest + @EnumSource(RuntimeExecutionMode.class) + void throwExceptionWithoutSettingUid(RuntimeExecutionMode runtimeExecutionMode) { + StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); + + final Configuration config = new Configuration(); + config.set(ExecutionOptions.RUNTIME_MODE, runtimeExecutionMode); + env.configure(config, getClass().getClassLoader()); + // disable auto generating uid + env.getConfig().disableAutoGeneratedUIDs(); + sinkTo(env.fromData(1, 2), simpleSink()); + assertThatThrownBy(env::getStreamGraph).isInstanceOf(IllegalStateException.class); + } + + @Test + void disableOperatorChain() { + StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); + + final DataStreamSource<Integer> src = env.fromData(1, 2); + final DataStreamSink<Integer> dataStreamSink = sinkTo(src, sinkWithCommitter()).name(NAME); + dataStreamSink.disableChaining(); + + final StreamGraph streamGraph = env.getStreamGraph(); + final StreamNode writer = findWriter(streamGraph); + final StreamNode committer = findCommitter(streamGraph); + + assertThat(writer.getOperatorFactory().getChainingStrategy()) + .isEqualTo(ChainingStrategy.NEVER); + assertThat(committer.getOperatorFactory().getChainingStrategy()) + .isEqualTo(ChainingStrategy.NEVER); + } + + void validateTopology( + StreamNode src, + Class<?> srcOutTypeInfo, + StreamNode dest, + Class<? extends StreamOperatorFactory> operatorFactoryClass, + int expectedParallelism, + int expectedMaxParallelism) { + + // verify src node + final StreamEdge srcOutEdge = src.getOutEdges().get(0); + assertThat(srcOutEdge.getTargetId()).isEqualTo(dest.getId()); + assertThat(src.getTypeSerializerOut()).isInstanceOf(srcOutTypeInfo); + + // verify dest node input + final StreamEdge destInputEdge = dest.getInEdges().get(0); + assertThat(destInputEdge.getTargetId()).isEqualTo(dest.getId()); + assertThat(dest.getTypeSerializersIn()[0]).isInstanceOf(srcOutTypeInfo); + + // make sure 2 sink operators have different names/uid + assertThat(dest.getOperatorName()).isNotEqualTo(src.getOperatorName()); + assertThat(dest.getTransformationUID()).isNotEqualTo(src.getTransformationUID()); + + assertThat(dest.getOperatorFactory()).isInstanceOf(operatorFactoryClass); + assertThat(dest.getParallelism()).isEqualTo(expectedParallelism); + assertThat(dest.getMaxParallelism()).isEqualTo(expectedMaxParallelism); + assertThat(dest.getOperatorFactory().getChainingStrategy()) + .isEqualTo(ChainingStrategy.ALWAYS); + assertThat(dest.getSlotSharingGroup()).isEqualTo(SLOT_SHARE_GROUP); + } + + StreamGraph buildGraph(Sink<Integer> sink, RuntimeExecutionMode runtimeExecutionMode) { + return buildGraph(sink, runtimeExecutionMode, true); + } + + StreamGraph buildGraph( + Sink<Integer> sink, + RuntimeExecutionMode runtimeExecutionMode, + boolean setSinkParallelism) { + StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); + + final Configuration config = new Configuration(); + config.set(ExecutionOptions.RUNTIME_MODE, runtimeExecutionMode); + env.configure(config, getClass().getClassLoader()); + final DataStreamSource<Integer> src = env.fromData(1, 2); + final DataStreamSink<Integer> dataStreamSink = sinkTo(src.rebalance(), sink); + setSinkProperty(dataStreamSink, setSinkParallelism); + // Trigger the plan generation but do not clear the transformations + env.getExecutionPlan(); + return env.getStreamGraph(); + } + + private void setSinkProperty( + DataStreamSink<Integer> dataStreamSink, boolean setSinkParallelism) { + dataStreamSink.name(NAME); + dataStreamSink.uid(UID); + if (setSinkParallelism) { + dataStreamSink.setParallelism(SinkV2TransformationTranslatorITCase.PARALLELISM); + } + dataStreamSink.slotSharingGroup(SLOT_SHARE_GROUP); + } + + StreamNode findNodeName(StreamGraph streamGraph, Predicate<String> predicate) { + return streamGraph.getStreamNodes().stream() + .filter(node -> predicate.test(node.getOperatorName())) + .findFirst() + .orElseThrow(() -> new IllegalStateException("Can not find the node")); + } + + StreamNode findWriter(StreamGraph streamGraph) { + return findNodeName( + streamGraph, name -> name.contains("Writer") && !name.contains("Committer")); + } + + StreamNode findCommitter(StreamGraph streamGraph) { + return findNodeName( + streamGraph, + name -> name.contains("Committer") && !name.contains("Global Committer")); + } + + StreamNode findGlobalCommitter(StreamGraph streamGraph) { + return findNodeName(streamGraph, name -> name.contains("Global Committer")); } } diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/TestSinkV2.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/TestSinkV2.java index fc1a9066c7c..18f934752a3 100644 --- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/TestSinkV2.java +++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/TestSinkV2.java @@ -37,6 +37,7 @@ import org.apache.flink.streaming.api.connector.sink2.CommittableMessage; import org.apache.flink.streaming.api.connector.sink2.CommittableMessageTypeInfo; import org.apache.flink.streaming.api.connector.sink2.CommittableSummary; import org.apache.flink.streaming.api.connector.sink2.CommittableWithLineage; +import org.apache.flink.streaming.api.connector.sink2.StandardSinkTopologies; import org.apache.flink.streaming.api.connector.sink2.SupportsPostCommitTopology; import org.apache.flink.streaming.api.connector.sink2.SupportsPreCommitTopology; import org.apache.flink.streaming.api.datastream.DataStream; @@ -241,7 +242,8 @@ public class TestSinkV2<InputT> implements Sink<InputT> { @Override public void addPostCommitTopology(DataStream<CommittableMessage<String>> committables) { - // We do not need to do anything for tests + StandardSinkTopologies.addGlobalCommitter( + committables, this::createCommitter, this::getCommittableSerializer); } }
