add freshMan readme dialog,datavane/tis#448

Author: baisui1981

tis-incr/tis-flink-extends/src/main/java/com/qlangtech/plugins/incr/flink/cdc/FlinkCDCPipelineEventProcess.java

@@ -62,6 +62,8 @@ public Object apply(Object o) {
         }
     }
 
+
+
     public static class FlinkPipelineDecimalConvert extends BiFunction {
         //  private final DataType type;
 

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

@@ -316,7 +316,9 @@ public FlinkCol varcharType(DataType type) {
                     //, DataTypes.VARCHAR(type.columnSize)
                     , new StringConvert()
                     , FlinkCol.NoOp()
-                    , new FlinkCDCPipelineEventProcess(org.apache.flink.cdc.common.types.DataTypes.VARCHAR(type.getColumnSize()), new FlinkPipelineStringConvert())
+                    , new FlinkCDCPipelineEventProcess(
+                    org.apache.flink.cdc.common.types.DataTypes.VARCHAR(type.getColumnSize())
+                    , new FlinkPipelineStringConvert())
                     , new RowFieldGetterFactory.StringGetter(meta.getName(), colIndex));
         }
 
@@ -336,9 +338,6 @@ public static FlinkCol mapFlinkCol(IColMetaGetter meta, int colIndex) {
     }
 
 
-
-
-
     protected abstract IMPLDATA createRowData(DTO dto);
 
     static class ShortConvert extends BiFunction {
@@ -385,6 +384,9 @@ public Object apply(Object o) {
             if (o instanceof String) {
                 return Integer.parseInt((String) o);
             }
+            if (o instanceof Number) {
+                return ((Number) o).intValue();
+            }
             return o;
         }
     }

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

@@ -19,10 +19,12 @@
 package com.qlangtech.tis.plugins.incr.flink.cdc;
 
 import com.alibaba.datax.common.element.ICol2Index;
-import org.apache.flink.calcite.shaded.com.google.common.collect.Maps;
+import com.qlangtech.plugins.incr.flink.cdc.FlinkCol;
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.lang3.tuple.Pair;
 
 import java.io.Serializable;
-import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.stream.Collectors;
@@ -34,8 +36,25 @@
 public class FlinkCol2Index implements ICol2Index, Serializable {
     final Map<String, ICol2Index.Col> col2IndexMapper;
 
+    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();
+        int idx = 0;
+        for (FlinkCol col : cols) {
+            col2IdxBuilder.put(col.name, Pair.of(idx++, col));
+        }
+
+        return new FlinkCol2Index(
+                col2IdxBuilder.entrySet().stream().collect(
+                        Collectors.toUnmodifiableMap((entry) -> entry.getKey() //
+                                , (entry) -> new ICol2Index.Col(entry.getValue().getKey(), entry.getValue().getValue().colType))));
+    }
+
+
     public FlinkCol2Index(Map<String, ICol2Index.Col> col2IndexMapper) {
-        this.col2IndexMapper = Maps.newHashMap(Objects.requireNonNull(col2IndexMapper, "param col2IndexMapper can not be null"));
+        this.col2IndexMapper = (Objects.requireNonNull(col2IndexMapper, "param col2IndexMapper can not be null"));
     }
 
     @Override

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

@@ -18,20 +18,15 @@
 
 package com.qlangtech.tis.plugins.incr.flink.cdc;
 
-import com.alibaba.datax.common.element.ICol2Index;
 import com.google.common.collect.Lists;
-import com.google.common.collect.Maps;
 import com.qlangtech.plugins.incr.flink.cdc.FlinkCol;
 import com.qlangtech.tis.plugin.datax.transformer.RecordTransformerRules;
 import com.qlangtech.tis.plugin.datax.transformer.UDFDefinition;
-import org.apache.commons.lang3.tuple.Pair;
 import org.apache.flink.api.common.functions.MapFunction;
 
 import java.io.Serializable;
 import java.util.List;
-import java.util.Map;
 import java.util.Objects;
-import java.util.stream.Collectors;
 
 /**
  * 执行TIS中定义的 transformer执行逻辑
@@ -45,21 +40,13 @@ public abstract class ReocrdTransformerMapper<Type> implements MapFunction<Type,
     private final FlinkCol2Index col2IdxMapper;
 
     public ReocrdTransformerMapper(List<FlinkCol> cols, RecordTransformerRules transformerRules) {
-        this.cols = Lists.newArrayList(Objects.requireNonNull(cols, "cols can not be null"));
-        this.transformerUDF
-                = Objects.requireNonNull(transformerRules, "param transformerRules can not be null")
-                .rules.stream().map((t) -> t.getUdf()).collect(Collectors.toList());
-
-        Map<String, Pair<Integer, FlinkCol>> col2IdxBuilder = Maps.newHashMap();
-        int idx = 0;
-        for (FlinkCol col : cols) {
-            col2IdxBuilder.put(col.name, Pair.of(idx++, col));
-        }
+        this(cols, transformerRules.getTransformerUDFs());
+    }
 
-        this.col2IdxMapper = new FlinkCol2Index(
-                col2IdxBuilder.entrySet().stream().collect(
-                        Collectors.toMap((entry) -> entry.getKey() //
-                                , (entry) -> new ICol2Index.Col(entry.getValue().getKey(), entry.getValue().getValue().colType))));
+    public ReocrdTransformerMapper(List<FlinkCol> cols, List<UDFDefinition> transformerUDF) {
+        this.cols = Lists.newArrayList(Objects.requireNonNull(cols, "cols can not be null"));
+        this.transformerUDF = transformerUDF;
+        this.col2IdxMapper = FlinkCol2Index.create(cols);
     }
 
     @Override