add method normalizeDecimalPrecisionAndScale()

Author: baisui1981

tis-datax/tis-ds-mysql-v5-plugin/src/main/java/com/qlangtech/tis/plugin/ds/mysql/MySQLV5DataSourceFactory.java

@@ -57,7 +57,10 @@ public JDBCConnection createConnection(String jdbcUrl, Optional<Properties> prop
         if (this.password != null) {
             info.put("password", this.password);
         }
-        info.put("serverTimezone", Objects.requireNonNull(this.timeZone, "timeZone can not be null").getTimeZone().getId());
+        if(this.timeZone != null){
+            info.put("serverTimezone", Objects.requireNonNull(this.timeZone, "timeZone can not be null").getTimeZone().getId());
+        }
+
         if (verify) {
             info.put("connectTimeout", "3000");
             info.put("socketTimeout", "3000");

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

@@ -219,12 +219,12 @@ public FlinkCol bigInt(DataType type) {
 
 
         public FlinkCol decimalType(DataType type) {
-
-            int precision = type.getColumnSize();
-            Integer scale = type.getDecimalDigits();
-            if (precision < 1 || precision > DataTypeMeta.DEFAULT_DECIMAL_PRECISION) {
-                precision = DataTypeMeta.DEFAULT_DECIMAL_PRECISION;
-            }
+            int[] normalize = DataTypeMeta.normalizeDecimalPrecisionAndScale(type);
+            int precision = normalize[0];// type.getColumnSize();
+            Integer scale = normalize[1];// type.getDecimalDigits();
+//            if (precision < 1 || precision > DataTypeMeta.DEFAULT_DECIMAL_PRECISION) {
+//                precision = DataTypeMeta.DEFAULT_DECIMAL_PRECISION;
+//            }
             try {
                 return new FlinkCol(meta, type, //DataTypes.DECIMAL(precision, scale)
                         new AtomicDataType(new DecimalType(nullable, precision, scale))