move AbstractTabSinkFuncV1 to tis-realtime-flink

Author: baisui1981

tis-incr/tis-flink-cdc-kafka-plugin/pom.xml

@@ -33,7 +33,7 @@
     <!--https://github.com/apache/flink-connector-kafka-->
     <!--https://nightlies.apache.org/flink/flink-docs-release-1.20/docs/connectors/datastream/kafka/-->
     <properties>
-        <flink-connector-kafka.version>3.2.0-1.18</flink-connector-kafka.version>
+        <flink-connector-kafka.version>3.4.0-1.20</flink-connector-kafka.version>
     </properties>
 
     <dependencies>

tis-incr/tis-flink-extends/pom.xml

@@ -30,7 +30,7 @@
 
     <artifactId>tis-flink-extends</artifactId>
     <properties>
-        <flink.connector-jdbc.version>3.1.2-1.18</flink.connector-jdbc.version>
+        <flink.connector-jdbc.version>3.3.0-1.20</flink.connector-jdbc.version>
     </properties>
 
     <dependencies>

tis-incr/tis-flink-extends/src/main/java/com/qlangtech/tis/realtime/BasicFlinkSourceHandle.java

@@ -106,15 +106,15 @@ public JobExecutionResult consume(boolean flinkCDCPipelineEnable, TargetResName
         }
 
         Tab2OutputTag<DTOStream> tab2OutputTag = createTab2OutputTag(asyncMsg, env, dataxName);
-        Map<TableAlias, AbstractTabSinkFuncV1<?, ?, SINK_TRANSFER_OBJ>> sinks = createTabSinkFunc(dataXProcessor);
+        Map<TableAlias, TabSinkFunc<?, ?, SINK_TRANSFER_OBJ>> sinks = createTabSinkFunc(dataXProcessor);
 
         this.processTableStream(env, tab2OutputTag, new SinkFuncs(sinks));
         return executeFlinkJob(dataxName, env);
     }
 
-    protected Map<TableAlias, AbstractTabSinkFuncV1<?, ?, SINK_TRANSFER_OBJ>> createTabSinkFunc(
+    protected Map<TableAlias, TabSinkFunc<?, ?, SINK_TRANSFER_OBJ>> createTabSinkFunc(
             IDataxProcessor dataXProcessor) {
-        Map<TableAlias, AbstractTabSinkFuncV1<?, ?, SINK_TRANSFER_OBJ>> sinks
+        Map<TableAlias, TabSinkFunc<?, ?, SINK_TRANSFER_OBJ>> sinks
                 = this.getSinkFuncFactory().createSinkFunction(dataXProcessor, flinkColCreator);
         sinks.forEach((tab, func) -> {
             if (StringUtils.isEmpty(tab.getTo()) || StringUtils.isEmpty(tab.getFrom())) {

…/tis/realtime/AbstractTabSinkFuncV1.java …/tis/realtime/AbstractTabSinkFuncV1.javatis-incr/tis-flink-extends/src/main/java/com/qlangtech/tis/realtime/AbstractTabSinkFuncV1.java renamed to tis-incr/tis-realtime-flink/src/main/java/com/qlangtech/tis/realtime/AbstractTabSinkFuncV1.java tis-incr/tis-flink-extends/src/main/java/com/qlangtech/tis/realtime/AbstractTabSinkFuncV1.java renamed to tis-incr/tis-realtime-flink/src/main/java/com/qlangtech/tis/realtime/AbstractTabSinkFuncV1.java

@@ -18,37 +18,14 @@
 
 package com.qlangtech.tis.realtime;
 
-import com.qlangtech.plugins.incr.flink.cdc.FlinkCol;
 import com.qlangtech.tis.async.message.client.consumer.IFlinkColCreator;
-import com.qlangtech.tis.datax.DataXName;
 import com.qlangtech.tis.datax.IDataxProcessor;
 import com.qlangtech.tis.datax.TableAlias;
-import com.qlangtech.tis.datax.impl.DataxProcessor;
-import com.qlangtech.tis.plugin.datax.transformer.RecordTransformerRules;
-import com.qlangtech.tis.plugin.ds.ISelectedTab;
 import com.qlangtech.tis.plugin.incr.TISSinkFactory;
-import com.qlangtech.tis.plugins.incr.flink.cdc.DTO2RowDataMapper;
-import com.qlangtech.tis.plugins.incr.flink.cdc.impl.RowDataTransformerMapper;
-import com.qlangtech.tis.realtime.dto.DTOStream;
-import com.qlangtech.tis.realtime.transfer.DTO;
-import com.qlangtech.tis.realtime.transfer.DTO.EventType;
-import com.qlangtech.tis.util.IPluginContext;
-import org.apache.commons.collections.CollectionUtils;
-import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.streaming.api.datastream.DataStream;
-import org.apache.flink.streaming.api.datastream.DataStreamSink;
-import org.apache.flink.streaming.api.functions.sink.SinkFunction;
-import org.apache.flink.table.connector.sink.SinkFunctionProvider;
-import org.apache.flink.table.data.RowData;
-import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
-import org.apache.flink.table.types.logical.LogicalType;
-import org.apache.flink.table.types.logical.RowType;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.List;
 import java.util.Map;
-import java.util.Optional;
 
 /**
  * @author: 百岁（baisui@qlangtech.com）

tis-incr/tis-realtime-flink/src/main/java/com/qlangtech/tis/realtime/BasicTISSinkFactory.java

@@ -106,7 +106,7 @@ protected final void processTableStream(StreamExecutionEnvironment env
     }
 
     @Override
-    protected Map<TableAlias, AbstractTabSinkFuncV1<?, ?, DTO>> createTabSinkFunc(IDataxProcessor dataXProcessor) {
+    protected Map<TableAlias, TabSinkFunc<?, ?, DTO>> createTabSinkFunc(IDataxProcessor dataXProcessor) {
         // return super.createTabSinkFunc(dataXProcessor);
         return Collections.emptyMap();
     }