[flink] 01/02: [hotfix][core] Complete the scala API for the new sink

[ga...@apache.org]

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

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

commit 46788bbabed543cbfff329b16b3f517231856c65
Author: Gen Luo <lu...@gmail.com>
AuthorDate: Wed Feb 9 15:29:52 2022 +0800

    [hotfix][core] Complete the scala API for the new sink
---
 .../org/apache/flink/streaming/api/scala/DataStream.scala    | 12 ++++++++++--
 1 file changed, 10 insertions(+), 2 deletions(-)

diff --git a/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/DataStream.scala b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/DataStream.scala
index f6a7bd9..2fd7522 100644
--- a/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/DataStream.scala
+++ b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/DataStream.scala
@@ -27,7 +27,7 @@ import org.apache.flink.api.common.operators.{ResourceSpec, SlotSharingGroup}
 import org.apache.flink.api.common.serialization.SerializationSchema
 import org.apache.flink.api.common.state.MapStateDescriptor
 import org.apache.flink.api.common.typeinfo.TypeInformation
-import org.apache.flink.api.connector.sink.Sink
+import org.apache.flink.api.connector.sink2.Sink
 import org.apache.flink.api.java.functions.KeySelector
 import org.apache.flink.api.java.tuple.{Tuple => JavaTuple}
 import org.apache.flink.api.java.typeutils.ResultTypeQueryable
@@ -1154,7 +1154,15 @@ class DataStream[T](stream: JavaStream[T]) {
    * will be executed once the StreamExecutionEnvironment.execute(...)
    * method is called.
    */
-  def sinkTo(sink: Sink[T, _, _, _]): DataStreamSink[T] = stream.sinkTo(sink)
+  def sinkTo(sink: org.apache.flink.api.connector.sink.Sink[T, _, _, _]): DataStreamSink[T] =
+    stream.sinkTo(sink)
+
+  /**
+   * Adds the given sink to this DataStream. Only streams with sinks added
+   * will be executed once the StreamExecutionEnvironment.execute(...)
+   * method is called.
+   */
+  def sinkTo(sink: Sink[T]): DataStreamSink[T] = stream.sinkTo(sink)
 
   /**
    * Triggers the distributed execution of the streaming dataflow and returns an iterator over the