add paimon sink plugin support

Author: baisui1981

tis-datax/tis-datax-common-plugin/src/main/java/com/qlangtech/tis/plugin/datax/common/BasicDataXRdbmsWriter.java

@@ -103,7 +103,7 @@ public abstract class BasicDataXRdbmsWriter<DS extends DataSourceFactory> extend
     @FormField(ordinal = 12, type = FormFieldType.INT_NUMBER, validate = {Validator.integer})
     public Integer batchSize;
 
-    @FormField(ordinal = 10, type = FormFieldType.ENUM, validate = {Validator.require})
+    @FormField(ordinal = 10, validate = {Validator.require})
     // 目标源中是否自动创建表，这样会方便不少
     public AutoCreateTable autoCreateTable;
 
@@ -228,7 +228,7 @@ public AutoCreateTable getAutoCreateTableCanNotBeNull() {
      * @return
      */
     @Override
-    public boolean isGenerateCreateDDLSwitchOff() {
+    public final boolean isGenerateCreateDDLSwitchOff() {
         return !getAutoCreateTableCanNotBeNull().enabled();
     }
 

tis-datax/tis-datax-odps-plugin/src/main/java/com/qlangtech/tis/plugin/datax/DataXOdpsWriter.java

@@ -197,10 +197,6 @@ public DataflowTask createTask(ISqlTask nodeMeta, boolean isFinalNode, IExecChai
         return odpsTask;
     }
 
-    @Override
-    public boolean isGenerateCreateDDLSwitchOff() {
-        return false;
-    }
 
     @Override
     public ExecuteResult startTask(ITableBuildTask dumpTask) {

tis-incr/tis-flink-extends/src/test/java/com/qlangtech/plugins/incr/flink/TestTISFlinkClassLoaderFactory.java

@@ -22,6 +22,7 @@
 import com.qlangtech.tis.async.message.client.consumer.IFlinkColCreator;
 import com.qlangtech.tis.async.message.client.consumer.impl.MQListenerFactory;
 import com.qlangtech.tis.coredefine.module.action.TargetResName;
+import com.qlangtech.tis.datax.StoreResourceTypeConstants;
 import com.qlangtech.tis.datax.TimeFormat;
 import com.qlangtech.tis.datax.impl.DataxProcessor;
 import com.qlangtech.tis.manage.common.CenterResource;
@@ -46,6 +47,7 @@
  **/
 public class TestTISFlinkClassLoaderFactory implements TISEasyMock {
 
+
     @Test
     public void testBuildServerLoaderFactory() throws Exception {
         BlobLibraryCacheManager.ClassLoaderFactory loaderFactory;

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

@@ -19,6 +19,8 @@
 package com.qlangtech.tis.plugins.incr.flink.cdc;
 
 import com.alibaba.datax.common.element.ICol2Index;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableMap.Builder;
 import com.qlangtech.plugins.incr.flink.cdc.FlinkCol;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.lang3.tuple.Pair;
@@ -40,16 +42,13 @@ public static FlinkCol2Index create(List<FlinkCol> cols) {
         if (CollectionUtils.isEmpty(cols)) {
             throw new IllegalArgumentException("param cols can not be empty");
         }
-        Map<String, Pair<Integer, FlinkCol>> col2IdxBuilder = com.google.common.collect.Maps.newHashMap();
+        Builder<String, ICol2Index.Col> col2IdxBuilder = ImmutableMap.builder();
         int idx = 0;
         for (FlinkCol col : cols) {
-            col2IdxBuilder.put(col.name, Pair.of(idx++, col));
+            col2IdxBuilder.put(col.name, new ICol2Index.Col(idx++, col.colType));
         }
 
-        return new FlinkCol2Index(
-                col2IdxBuilder.entrySet().stream().collect(
-                        Collectors.toUnmodifiableMap((entry) -> entry.getKey() //
-                                , (entry) -> new ICol2Index.Col(entry.getValue().getKey(), entry.getValue().getValue().colType))));
+        return new FlinkCol2Index(col2IdxBuilder.build());
     }