(flink-cdc) branch master updated: [FLINK-35149][cdc-composer] Fix DataSinkTranslator#sinkTo ignoring pre-write topology if not TwoPhaseCommittingSink (#3233)

[re...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

renqs pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git


The following commit(s) were added to refs/heads/master by this push:
     new 33891869a [FLINK-35149][cdc-composer] Fix DataSinkTranslator#sinkTo ignoring pre-write topology if not TwoPhaseCommittingSink (#3233)
33891869a is described below

commit 33891869a9fffa2abf8b8ae03915d0ddccdaf5ec
Author: Hongshun Wang <12...@users.noreply.github.com>
AuthorDate: Tue Jun 4 16:39:50 2024 +0800

    [FLINK-35149][cdc-composer] Fix DataSinkTranslator#sinkTo ignoring pre-write topology if not TwoPhaseCommittingSink (#3233)
---
 .../flink/translator/DataSinkTranslator.java       |  6 +-
 .../flink/translator/DataSinkTranslatorTest.java   | 89 ++++++++++++++++++++++
 2 files changed, 93 insertions(+), 2 deletions(-)

diff --git a/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/DataSinkTranslator.java b/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/DataSinkTranslator.java
index ec165dd56..a6188d2df 100644
--- a/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/DataSinkTranslator.java
+++ b/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/DataSinkTranslator.java
@@ -21,6 +21,7 @@ import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.connector.sink2.Sink;
 import org.apache.flink.api.connector.sink2.TwoPhaseCommittingSink;
 import org.apache.flink.cdc.common.annotation.Internal;
+import org.apache.flink.cdc.common.annotation.VisibleForTesting;
 import org.apache.flink.cdc.common.event.Event;
 import org.apache.flink.cdc.common.sink.DataSink;
 import org.apache.flink.cdc.common.sink.EventSinkProvider;
@@ -71,7 +72,8 @@ public class DataSinkTranslator {
         }
     }
 
-    private void sinkTo(
+    @VisibleForTesting
+    void sinkTo(
             DataStream<Event> input,
             Sink<Event> sink,
             String sinkName,
@@ -85,7 +87,7 @@ public class DataSinkTranslator {
         if (sink instanceof TwoPhaseCommittingSink) {
             addCommittingTopology(sink, stream, sinkName, schemaOperatorID);
         } else {
-            input.transform(
+            stream.transform(
                     SINK_WRITER_PREFIX + sinkName,
                     CommittableMessageTypeInfo.noOutput(),
                     new DataSinkWriterOperatorFactory<>(sink, schemaOperatorID));
diff --git a/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/translator/DataSinkTranslatorTest.java b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/translator/DataSinkTranslatorTest.java
new file mode 100644
index 000000000..1970f3dfa
--- /dev/null
+++ b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/translator/DataSinkTranslatorTest.java
@@ -0,0 +1,89 @@
+/*
+ * 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.cdc.composer.flink.translator;
+
+import org.apache.flink.api.connector.sink2.SinkWriter;
+import org.apache.flink.api.dag.Transformation;
+import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.streaming.api.connector.sink2.WithPreWriteTopology;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.transformations.OneInputTransformation;
+
+import org.apache.flink.shaded.guava31.com.google.common.collect.Lists;
+
+import org.junit.Assert;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.ArrayList;
+
+/** A test for {@link DataSinkTranslator}. */
+public class DataSinkTranslatorTest {
+
+    @Test
+    public void testPreWriteWithoutCommitSink() {
+        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+        ArrayList<Event> mockEvents = Lists.newArrayList(new EmptyEvent(), new EmptyEvent());
+        DataStreamSource<Event> inputStream = env.fromCollection(mockEvents);
+        DataSinkTranslator translator = new DataSinkTranslator();
+
+        // Node hash must be a 32 character String that describes a hex code
+        String uid = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
+        MockPreWriteWithoutCommitSink mockPreWriteWithoutCommitSink =
+                new MockPreWriteWithoutCommitSink(uid);
+        translator.sinkTo(
+                inputStream,
+                mockPreWriteWithoutCommitSink,
+                "testPreWriteWithoutCommitSink",
+                new OperatorID());
+
+        // Check if the `addPreWriteTopology` is called, and the uid is set when the transformation
+        // added
+        OneInputTransformation<Event, Event> oneInputTransformation =
+                (OneInputTransformation) env.getTransformations().get(0);
+        Transformation<?> reblanceTransformation = oneInputTransformation.getInputs().get(0);
+        Assert.assertEquals(uid, reblanceTransformation.getUserProvidedNodeHash());
+    }
+
+    private static class EmptyEvent implements Event {}
+
+    private static class MockPreWriteWithoutCommitSink implements WithPreWriteTopology<Event> {
+
+        private final String uid;
+
+        public MockPreWriteWithoutCommitSink(String uid) {
+            this.uid = uid;
+        }
+
+        @Override
+        public DataStream<Event> addPreWriteTopology(DataStream<Event> inputDataStream) {
+            // return a new DataSteam with specified uid
+            DataStream<Event> rebalance = inputDataStream.rebalance();
+            rebalance.getTransformation().setUidHash(uid);
+            return rebalance;
+        }
+
+        @Override
+        public SinkWriter<Event> createWriter(InitContext context) throws IOException {
+            return null;
+        }
+    }
+}