[FLINK-34548][API] Supports sink-v2 Sink

Author: reswqa

flink-core-api/src/main/java/org/apache/flink/api/connector/v2/Sink.java

@@ -0,0 +1,28 @@
+/*
+ * 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.api.connector.v2;
+
+import org.apache.flink.annotation.Experimental;
+
+/**
+ * Sink interface for DataStream api v2. Note that this is a temporary approach, will be removed in
+ * the future.
+ */
+@Experimental
+public interface Sink<T> {}

flink-core/src/main/java/org/apache/flink/api/connector/v2/SinkUtils.java

@@ -0,0 +1,29 @@
+/*
+ * 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.api.connector.v2;
+
+import org.apache.flink.annotation.Experimental;
+
+/** Utils to convert the sink-v2 based sink to a DataStream v2 Source. */
+@Experimental
+public class SinkUtils {
+    public static <T> Sink<T> wrapSink(org.apache.flink.api.connector.sink2.Sink<T> sink) {
+        return new WrappedSink<>(sink);
+    }
+}

flink-core/src/main/java/org/apache/flink/api/connector/v2/WrappedSink.java

@@ -0,0 +1,35 @@
+/*
+ * 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.api.connector.v2;
+
+import org.apache.flink.annotation.Internal;
+
+/** A simple {@link Sink} implementation that wrap a sink-v2 based sink. */
+@Internal
+public class WrappedSink<T> implements Sink<T> {
+    org.apache.flink.api.connector.sink2.Sink<T> wrappedSink;
+
+    public WrappedSink(org.apache.flink.api.connector.sink2.Sink<T> wrappedSink) {
+        this.wrappedSink = wrappedSink;
+    }
+
+    public org.apache.flink.api.connector.sink2.Sink<T> getWrappedSink() {
+        return wrappedSink;
+    }
+}

flink-process-function-parent/flink-process-function-api/src/main/java/org/apache/flink/process/api/stream/GlobalStream.java

@@ -19,6 +19,7 @@
 package org.apache.flink.process.api.stream;
 
 import org.apache.flink.annotation.Experimental;
+import org.apache.flink.api.connector.v2.Sink;
 import org.apache.flink.api.java.functions.KeySelector;
 import org.apache.flink.process.api.function.OneInputStreamProcessFunction;
 import org.apache.flink.process.api.function.TwoInputNonBroadcastStreamProcessFunction;
@@ -78,7 +79,7 @@ <T_OTHER, OUT> GlobalStream<OUT> connectAndProcess(
      */
     BroadcastStream<T> broadcast();
 
-    // TODO add toSink method.
+    void toSink(Sink<T> sink);
 
     /**
      * This class represents a combination of two {@link GlobalStream}. It will be used as the

flink-process-function-parent/flink-process-function-api/src/main/java/org/apache/flink/process/api/stream/KeyedPartitionStream.java

@@ -19,6 +19,7 @@
 package org.apache.flink.process.api.stream;
 
 import org.apache.flink.annotation.Experimental;
+import org.apache.flink.api.connector.v2.Sink;
 import org.apache.flink.api.java.functions.KeySelector;
 import org.apache.flink.process.api.function.OneInputStreamProcessFunction;
 import org.apache.flink.process.api.function.TwoInputBroadcastStreamProcessFunction;
@@ -168,7 +169,7 @@ <T_OTHER, OUT> KeyedPartitionStream<K, OUT> connectAndProcess(
      */
     BroadcastStream<T> broadcast();
 
-    // TODO add toSink method.
+    void toSink(Sink<T> sink);
 
     /**
      * This class represents a combination of two {@link KeyedPartitionStream}. It will be used as

flink-process-function-parent/flink-process-function-api/src/main/java/org/apache/flink/process/api/stream/NonKeyedPartitionStream.java

@@ -19,6 +19,7 @@
 package org.apache.flink.process.api.stream;
 
 import org.apache.flink.annotation.Experimental;
+import org.apache.flink.api.connector.v2.Sink;
 import org.apache.flink.api.java.functions.KeySelector;
 import org.apache.flink.process.api.function.OneInputStreamProcessFunction;
 import org.apache.flink.process.api.function.TwoInputBroadcastStreamProcessFunction;
@@ -100,7 +101,7 @@ <T_OTHER, OUT> NonKeyedPartitionStream<OUT> connectAndProcess(
      */
     BroadcastStream<T> broadcast();
 
-    // TODO add toSink method.
+    void toSink(Sink<T> sink);
 
     /**
      * This interface represents a combination of two {@link NonKeyedPartitionStream}. It will be

flink-process-function-parent/flink-process-function/src/main/java/org/apache/flink/process/impl/ExecutionEnvironmentImpl.java

@@ -48,6 +48,7 @@
 import org.apache.flink.streaming.api.graph.StreamGraph;
 import org.apache.flink.streaming.api.graph.StreamGraphGenerator;
 import org.apache.flink.streaming.api.transformations.SourceTransformation;
+import org.apache.flink.streaming.runtime.translators.ProcessFunctionSinkTransformationTranslator;
 import org.apache.flink.util.ExceptionUtils;
 import org.apache.flink.util.FlinkException;
 import org.apache.flink.util.Preconditions;
@@ -76,6 +77,18 @@ public class ExecutionEnvironmentImpl implements ExecutionEnvironment {
 
     private final PipelineExecutorServiceLoader executorServiceLoader;
 
+    static {
+        try {
+            // All transformation translator must be put to a map in StreamGraphGenerator, but
+            // streaming-java is not depend on process-function module, using reflect to handle
+            // this.
+            ProcessFunctionSinkTransformationTranslator.registerSinkTransformationTranslator();
+        } catch (Exception e) {
+            throw new RuntimeException(
+                    "Can not register process function transformation translator.");
+        }
+    }
+
     /**
      * The environment of the context (local by default, cluster if invoked through command line).
      */
@@ -180,6 +193,14 @@ public List<Transformation<?>> getTransformations() {
         return transformations;
     }
 
+    public void setParallelism(int parallelism) {
+        executionConfig.setParallelism(parallelism);
+    }
+
+    public CheckpointConfig getCheckpointCfg() {
+        return checkpointCfg;
+    }
+
     // -----------------------------------------------
     //              Internal Methods
     // -----------------------------------------------

flink-process-function-parent/flink-process-function/src/main/java/org/apache/flink/process/impl/stream/GlobalStreamImpl.java

@@ -19,6 +19,7 @@
 package org.apache.flink.process.impl.stream;
 
 import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.v2.Sink;
 import org.apache.flink.api.dag.Transformation;
 import org.apache.flink.api.java.functions.KeySelector;
 import org.apache.flink.api.java.tuple.Tuple2;
@@ -38,6 +39,7 @@
 import org.apache.flink.streaming.api.operators.SimpleUdfStreamOperatorFactory;
 import org.apache.flink.streaming.api.transformations.OneInputTransformation;
 import org.apache.flink.streaming.api.transformations.PartitionTransformation;
+import org.apache.flink.streaming.api.transformations.ProcessFunctionSinkTransformation;
 import org.apache.flink.streaming.runtime.partitioner.ShufflePartitioner;
 import org.apache.flink.util.OutputTag;
 
@@ -98,7 +100,14 @@ public <T_OTHER, OUT> GlobalStream<OUT> connectAndProcess(
         return new GlobalStreamImpl<>(environment, outTransformation);
     }
 
-    // TODO add toSink method.
+    @Override
+    public void toSink(Sink<T> sink) {
+        ProcessFunctionSinkTransformation<T, T> sinkTransformation =
+                StreamUtils.addSinkOperator(this, sink, getType());
+        // Operator parallelism should always be 1 for global stream.
+        // parallelismConfigured should be true to avoid overwritten by AdaptiveBatchScheduler.
+        sinkTransformation.setParallelism(1, true);
+    }
 
     // ---------------------
     //   Partitioning

flink-process-function-parent/flink-process-function/src/main/java/org/apache/flink/process/impl/stream/KeyedPartitionStreamImpl.java

@@ -19,6 +19,7 @@
 package org.apache.flink.process.impl.stream;
 
 import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.v2.Sink;
 import org.apache.flink.api.dag.Transformation;
 import org.apache.flink.api.java.functions.KeySelector;
 import org.apache.flink.api.java.tuple.Tuple2;
@@ -322,7 +323,10 @@ public KeySelector<V, K> getKeySelector() {
         return keySelector;
     }
 
-    // TODO add toSink method.
+    @Override
+    public void toSink(Sink<V> sink) {
+        StreamUtils.addSinkOperator(this, sink, getType());
+    }
 
     // ---------------------
     //   Partitioning

flink-process-function-parent/flink-process-function/src/main/java/org/apache/flink/process/impl/stream/NonKeyedPartitionStreamImpl.java

@@ -19,6 +19,7 @@
 package org.apache.flink.process.impl.stream;
 
 import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.v2.Sink;
 import org.apache.flink.api.dag.Transformation;
 import org.apache.flink.api.java.functions.KeySelector;
 import org.apache.flink.api.java.tuple.Tuple2;
@@ -132,7 +133,10 @@ public <T_OTHER, OUT> NonKeyedPartitionStream<OUT> connectAndProcess(
         return new NonKeyedPartitionStreamImpl<>(environment, outTransformation);
     }
 
-    // TODO add toSink method.
+    @Override
+    public void toSink(Sink<T> sink) {
+        StreamUtils.addSinkOperator(this, sink, getType());
+    }
 
     // ---------------------
     //   Partitioning

flink-process-function-parent/flink-process-function/src/main/java/org/apache/flink/process/impl/utils/StreamUtils.java

@@ -19,6 +19,8 @@
 package org.apache.flink.process.impl.utils;
 
 import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.v2.Sink;
+import org.apache.flink.api.connector.v2.WrappedSink;
 import org.apache.flink.api.java.Utils;
 import org.apache.flink.api.java.functions.KeySelector;
 import org.apache.flink.api.java.tuple.Tuple2;
@@ -29,11 +31,14 @@
 import org.apache.flink.process.api.function.TwoOutputStreamProcessFunction;
 import org.apache.flink.process.impl.stream.DataStream;
 import org.apache.flink.process.impl.stream.KeyedPartitionStreamImpl;
+import org.apache.flink.process.impl.stream.NonKeyedPartitionStreamImpl;
 import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
 import org.apache.flink.streaming.api.operators.SimpleOperatorFactory;
 import org.apache.flink.streaming.api.operators.SimpleUdfStreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.StreamOperatorFactory;
 import org.apache.flink.streaming.api.operators.TwoInputStreamOperator;
 import org.apache.flink.streaming.api.transformations.OneInputTransformation;
+import org.apache.flink.streaming.api.transformations.ProcessFunctionSinkTransformation;
 import org.apache.flink.streaming.api.transformations.TwoInputTransformation;
 
 /** Utils to handle things like extract type information for DataStream. */
@@ -225,4 +230,56 @@ public static <IN1, IN2, OUT> TwoInputTransformation<IN1, IN2, OUT> getTwoInputT
 
         return transform;
     }
+
+    /** Construct and return a new DataStream with one input operator. */
+    public static <T, R> DataStream<R> transformOneInputOperator(
+            String operatorName,
+            DataStream<T> inputStream,
+            TypeInformation<R> outTypeInfo,
+            StreamOperatorFactory<R> operatorFactory) {
+        // read the output type of the input Transform to coax out errors about MissingTypeInfo
+        inputStream.getTransformation().getOutputType();
+
+        OneInputTransformation<T, R> resultTransform =
+                new OneInputTransformation<>(
+                        inputStream.getTransformation(),
+                        operatorName,
+                        operatorFactory,
+                        outTypeInfo,
+                        inputStream.getEnvironment().getParallelism(),
+                        false);
+
+        NonKeyedPartitionStreamImpl<R> returnStream =
+                new NonKeyedPartitionStreamImpl<>(inputStream.getEnvironment(), resultTransform);
+
+        inputStream.getEnvironment().addOperator(resultTransform);
+
+        return returnStream;
+    }
+
+    /** Add sink operator to the input stream. */
+    public static <T> ProcessFunctionSinkTransformation<T, T> addSinkOperator(
+            DataStream<T> inputStream, Sink<T> sink, TypeInformation<T> typeInformation) {
+        // read the output type of the input Transform to coax out errors about MissingTypeInfo
+        inputStream.getTransformation().getOutputType();
+
+        if (!(sink instanceof WrappedSink)) {
+            throw new UnsupportedOperationException(
+                    "Unsupported type of sink, please use SinkUtils to wrap a sink-v2 sink first.");
+        }
+
+        org.apache.flink.api.connector.sink2.Sink<T> innerSink =
+                ((WrappedSink<T>) sink).getWrappedSink();
+
+        ProcessFunctionSinkTransformation<T, T> sinkTransformation =
+                new ProcessFunctionSinkTransformation<>(
+                        inputStream,
+                        innerSink,
+                        typeInformation,
+                        "Sink",
+                        inputStream.getEnvironment().getParallelism(),
+                        false);
+        inputStream.getEnvironment().addOperator(sinkTransformation);
+        return sinkTransformation;
+    }
 }