enable flink-cdc pipeline mode

Author: baisui1981

tis-incr/tis-chunjun-base-plugin/src/main/java/com/qlangtech/tis/plugins/incr/flink/chunjun/source/ChunjunSourceFunction.java

@@ -126,7 +126,7 @@ public AsyncMsg<List<ReaderSource>> start(boolean flinkCDCPipelineEnable, Target
 
 
         try {
-            SourceChannel sourceChannel = new SourceChannel(sourceFuncs);
+            SourceChannel sourceChannel = new SourceChannel(false, sourceFuncs);
             sourceChannel.setFocusTabs(tabs, dataXProcessor.getTabAlias(null), (tabName) -> DTOStream.createRowData());
             return sourceChannel;
             // return (JobExecutionResult) getConsumerHandle().consume(false, name, sourceChannel, dataXProcessor);

tis-incr/tis-flink-cdc-kafka-plugin/src/main/java/com/qlangtech/tis/plugin/kafka/consumer/FlinkKafkaFunction.java

@@ -84,7 +84,7 @@ public AsyncMsg<List<ReaderSource>> start(
 
         KafkaSource<DTO> source = kafkaSourceBuilder.build();
         try {
-            SourceChannel sourceChannel = new SourceChannel(
+            SourceChannel sourceChannel = new SourceChannel(flinkCDCPipelineEnable,
                     createKafkaSource(kafkaReader.bootstrapServers, source));
 
             sourceChannel.setFocusTabs(tabs, dataXProcessor.getTabAlias(null)

tis-incr/tis-flink-cdc-kingbase-plugin/src/main/java/com/qlangtech/plugins/incr/flink/cdc/pglike/FlinkCDCPGLikeSourceFunction.java

@@ -110,7 +110,7 @@ public AsyncMsg<List<ReaderSource>> start(boolean flinkCDCPipelineEnable, Target
                     });
 
 
-            SourceChannel sourceChannel = new SourceChannel(readerSources);
+            SourceChannel sourceChannel = new SourceChannel(flinkCDCPipelineEnable,readerSources);
             // for (ISelectedTab tab : tabs) {
             sourceChannel.setFocusTabs(tabs, dataXProcessor.getTabAlias(null), DTOStream::createDispatched);
             //}

tis-incr/tis-flink-cdc-mongdb-plugin/src/main/java/com/qlangtech/plugins/incr/flink/cdc/mongdb/FlinkCDCMongoDBSourceFunction.java

@@ -99,7 +99,7 @@ public AsyncMsg<List<ReaderSource>> start(boolean flinkCDCPipelineEnable, Target
                     , new DefaultTableNameConvert()
                     , contextParamValsGetterMapper);
 
-            SourceChannel sourceChannel = new SourceChannel(
+            SourceChannel sourceChannel = new SourceChannel(flinkCDCPipelineEnable,
                     SourceChannel.getSourceFunction(dsFactory, tabs, (dbHost, dbs, tbs, debeziumProperties) -> {
                         List<ReaderSource> sourceFunctions = createSourceFunctions(dsFactory, tabs, deserializationSchema);
                         return sourceFunctions;

tis-incr/tis-flink-cdc-mysql-plugin/src/main/java/com/qlangtech/tis/plugins/incr/flink/cdc/mysql/FlinkCDCMysqlSourceFunction.java

@@ -214,6 +214,7 @@ public AsyncMsg<List<ReaderSource>> start(boolean flinkCDCPipelineEnable, Target
 
 
             SourceChannel sourceChannel = new SourceChannel(
+                    flinkCDCPipelineEnable,
                     SourceChannel.getSourceFunction(
                             dsFactory,
                             tabs

tis-incr/tis-flink-cdc-oracle-plugin/src/main/java/com/qlangtech/plugins/incr/flink/cdc/oracle/FlinkCDCOracleSourceFunction.java

@@ -108,7 +108,7 @@ public AsyncMsg<List<ReaderSource>> start(boolean flinkCDCPipelineEnable, Target
                     , tablesInDB.getPhysicsTabName2LogicNameConvertor()
                     , contextParamValsGetterMapper);
 
-            SourceChannel sourceChannel = new SourceChannel(
+            SourceChannel sourceChannel = new SourceChannel(flinkCDCPipelineEnable,
                     SourceChannel.getSourceFunction(dsFactory, tabs
                             , (dbHost, dbs, tbs, debeziumProperties) -> {
                                 return dbs.getDbStream().map((databaseName) -> {

tis-incr/tis-flink-cdc-postgresql-plugin/src/main/java/com/qlangtech/plugins/incr/flink/cdc/pglike/FlinkCDCPGLikeSourceFunction.java

@@ -112,7 +112,7 @@ public AsyncMsg<List<ReaderSource>> start(
                     });
 
 
-            SourceChannel sourceChannel = new SourceChannel(readerSources);
+            SourceChannel sourceChannel = new SourceChannel(flinkCDCPipelineEnable, readerSources);
             // for (ISelectedTab tab : tabs) {
             sourceChannel.setFocusTabs(tabs, dataXProcessor.getTabAlias(null), DTOStream::createDispatched);
             //}

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

@@ -18,21 +18,37 @@
 
 package com.qlangtech.tis.realtime;
 
+import com.qlangtech.tis.plugin.ds.ISelectedTab;
 import com.qlangtech.tis.realtime.transfer.DTO;
 import org.apache.flink.util.OutputTag;
 
-import java.util.Collections;
+import java.util.List;
 import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 /**
+ *
  * @author: 百岁（baisui@qlangtech.com）
  * @create: 2025-01-04 09:22
  **/
 public class DTOSourceTagProcessFunction extends SourceProcessFunction<DTO> {
     public static final String KEY_MERGE_ALL_TABS_IN_ONE_BUS = "merge_all_tabs_in_one_bus";
 
+    public static Set<String> createFocusTabs(boolean flinkCDCPipelineEnable, List<ISelectedTab> tabs) {
+        return (flinkCDCPipelineEnable
+                ? Stream.of(DTOSourceTagProcessFunction.KEY_MERGE_ALL_TABS_IN_ONE_BUS)
+                : tabs.stream().map((t) -> t.getName())).collect(Collectors.toSet());
+    }
+
     public static DTOSourceTagProcessFunction create(boolean flinkCDCPipelineEnable, Map<String, OutputTag<DTO>> tab2OutputTag) {
-        return flinkCDCPipelineEnable ? new MergeAllTabsInOneBusProcessFunction() : new DTOSourceTagProcessFunction(tab2OutputTag);
+        if (flinkCDCPipelineEnable) {
+            if (tab2OutputTag.size() != 1 || !tab2OutputTag.containsKey(KEY_MERGE_ALL_TABS_IN_ONE_BUS)) {
+                throw new IllegalStateException("the size of tab2OutputTag must be 1,but now is:" + String.join(",", tab2OutputTag.keySet()));
+            }
+        }
+        return flinkCDCPipelineEnable ? new MergeAllTabsInOneBusProcessFunction(tab2OutputTag) : new DTOSourceTagProcessFunction(tab2OutputTag);
     }
 
     public DTOSourceTagProcessFunction(Map<String, OutputTag<DTO>> tab2OutputTag) {
@@ -47,9 +63,8 @@ protected String getTableName(DTO record) {
 
     static class MergeAllTabsInOneBusProcessFunction extends DTOSourceTagProcessFunction {
 
-        private MergeAllTabsInOneBusProcessFunction() {
-            super(Collections.singletonMap(KEY_MERGE_ALL_TABS_IN_ONE_BUS, new OutputTag<DTO>(KEY_MERGE_ALL_TABS_IN_ONE_BUS) {
-            }));
+        public MergeAllTabsInOneBusProcessFunction(Map<String, OutputTag<DTO>> tab2OutputTag) {
+            super(tab2OutputTag);
         }
 
         @Override

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

@@ -25,6 +25,8 @@
 import java.util.Map;
 
 /**
+ * 负责给主数据流打标，提供给后续流程 {@link com.qlangtech.tis.realtime.dto.DTOStream.DispatchedDTOStream} 实现分流
+ *
  * @author: 百岁（baisui@qlangtech.com）
  * @create: 2021-10-27 10:38
  **/
@@ -35,6 +37,17 @@ public SourceProcessFunction(Map<String, OutputTag<RECORD_TYPE>> tab2OutputTag)
         this.tab2OutputTag = tab2OutputTag;
     }
 
+    /**
+     * 在主流中为每个表打标签
+     *
+     * @param in   The input value.
+     * @param ctx  A {@link Context} that allows querying the timestamp of the element and getting a
+     *             {@link org.apache.flink.streaming.api.TimerService} for registering timers and querying the time. The context is only
+     *             valid during the invocation of this method, do not store it.
+     * @param _out The collector for returning result values.
+     * @throws Exception
+     * @see com.qlangtech.tis.realtime.dto.DTOStream.DispatchedDTOStream#addStream collected in it
+     */
     @Override
     public void processElement(RECORD_TYPE in, Context ctx, Collector<RECORD_TYPE> _out) throws Exception {
         //side_output: https://ci.apache.org/projects/flink/flink-docs-stable/dev/stream/side_output.html

tis-incr/tis-realtime-flink/src/main/java/com/qlangtech/plugins/incr/flink/cdc/SourceChannel.java

@@ -32,6 +32,7 @@
 import com.qlangtech.tis.plugin.ds.DataSourceFactory;
 import com.qlangtech.tis.plugin.ds.ISelectedTab;
 import com.qlangtech.tis.plugin.ds.TableInDB;
+import com.qlangtech.tis.realtime.DTOSourceTagProcessFunction;
 import com.qlangtech.tis.realtime.ReaderSource;
 import com.qlangtech.tis.realtime.dto.DTOStream;
 import com.qlangtech.tis.sql.parser.tuple.creator.EntityName;
@@ -59,21 +60,23 @@ public class SourceChannel implements AsyncMsg<List<ReaderSource>> {
     private final List<ReaderSource> sourceFunction;
     private Set<String> focusTabs = null;// = Sets.newHashSet();
     private Tab2OutputTag<DTOStream> tab2OutputTag = null;
+    private final boolean flinkCDCPipelineEnable;
 
     @Override
     public Tab2OutputTag<DTOStream> getTab2OutputTag() {
         return Objects.requireNonNull(tab2OutputTag);
     }
 
-    public SourceChannel(List<ReaderSource> sourceFunction) {
+    public SourceChannel(boolean flinkCDCPipelineEnable, List<ReaderSource> sourceFunction) {
         if (CollectionUtils.isEmpty(sourceFunction)) {
             throw new IllegalArgumentException("param sourceFunction can not be empty");
         }
         this.sourceFunction = sourceFunction;
+        this.flinkCDCPipelineEnable = flinkCDCPipelineEnable;
     }
 
-    public SourceChannel(ReaderSource sourceFunction) {
-        this(Collections.singletonList(sourceFunction));
+    public SourceChannel(boolean flinkCDCPipelineEnable, ReaderSource sourceFunction) {
+        this(flinkCDCPipelineEnable, Collections.singletonList(sourceFunction));
     }
 
     public static List<ReaderSource> getSourceFunction(
@@ -170,12 +173,15 @@ public void setFocusTabs(List<ISelectedTab> tabs, TableAliasMapper tabAliasMappe
         if (tabAliasMapper.isNull()) {
             throw new IllegalArgumentException("param tabAliasMapper can not be null");
         }
-        this.focusTabs = tabs.stream().map((t) -> t.getName()).collect(Collectors.toSet());
+        this.focusTabs = DTOSourceTagProcessFunction.createFocusTabs(this.flinkCDCPipelineEnable, tabs);
+//     (this.flinkCDCPipelineEnable
+//                ? Stream.of(DTOSourceTagProcessFunction.KEY_MERGE_ALL_TABS_IN_ONE_BUS)
+//                : tabs.stream().map((t) -> t.getName())).collect(Collectors.toSet());
 
-        Map<TableAlias, DTOStream> tab2StreamMapper = tabs.stream().collect(
+        Map<TableAlias, DTOStream> tab2StreamMapper = this.focusTabs.stream().collect(
                 Collectors.toMap(
-                        (tab) -> (tabAliasMapper.getWithCheckNotNull(tab.getName()))
-                        , (t) -> dtoStreamCreator.apply(t.getName())));
+                        (name) -> (tabAliasMapper.getWithCheckNotNull(name))
+                        , (name) -> dtoStreamCreator.apply(name)));
         this.tab2OutputTag
                 = new Tab2OutputTag<>(tab2StreamMapper);
     }