Fix 0.6.7-rc1 (#1017)

* [Fix][admin,web,app,core] Fix rc1 bugs (#1007)
Co-authored-by: wenmo <32723967+wenmo@users.noreply.github.com>

* alter error (#1011)

* [Optimization-1014][client] Optimizate Doris sink and type convert and upgrade flink to 1.15.2 (#1015)
Co-authored-by: wenmo <32723967+wenmo@users.noreply.github.com>

* UI page display misalignment fix (#1013)

Co-authored-by: steve <woai1998>

* out.collect在字段报错时没有上级抛错误,e.getCause().getMessage()会打不出错误栈，打印修改为logger.error("..",e). (#1016)
Co-authored-by: wenmo <32723967+wenmo@users.noreply.github.com>
Co-authored-by: toms <94617906+Toms1999@users.noreply.github.com>
Co-authored-by: ikiler <34333571+ikiler@users.noreply.github.com>
Co-authored-by: mengyejiang <403905717@qq.com>

dlink-admin/src/main/java/com/dlink/controller/FileUploadController.java

@@ -72,4 +72,19 @@ public class FileUploadController {
         }
     }
 
+    /**
+     * Upload hdfs file<br>
+     *
+     * @param files            Multi files
+     * @param dir              Dir, default is empty. If not provide, please provide the 'fileType' value
+     * @param hadoopConfigPath Please refer {@link UploadFileConstant}, default is -1. If not provide, please provide the 'dir' value
+     * @return {@link Result}
+     */
+    @PostMapping(value = "hdfs")
+    public Result uploadHdfs(@RequestPart("files") MultipartFile[] files,
+                             @RequestParam(value = "dir", defaultValue = "", required = false) String dir,
+                             @RequestParam(value = "hadoopConfigPath", required = false) String hadoopConfigPath) {
+        return fileUploadService.uploadHdfs(files, dir, hadoopConfigPath);
+    }
+
 }

dlink-admin/src/main/java/com/dlink/service/FileUploadService.java

@@ -34,7 +34,7 @@ public interface FileUploadService {
      *
      * @param file     {@link MultipartFile} instance
      * @param fileType Upload file's type, refer ${@link UploadFileConstant}
-     * @return {@link com.dlink.common.result.Result}
+     * @return {@link Result}
      */
     Result upload(MultipartFile file, Byte fileType);
 
@@ -43,7 +43,7 @@ public interface FileUploadService {
      *
      * @param files    {@link MultipartFile} instance
      * @param fileType Upload file's type, refer ${@link UploadFileConstant}
-     * @return {@link com.dlink.common.result.Result}
+     * @return {@link Result}
      */
     Result upload(MultipartFile[] files, Byte fileType);
 
@@ -53,7 +53,7 @@ public interface FileUploadService {
      * @param file     {@link MultipartFile} instance
      * @param dir      Local absolute dir
      * @param fileType Upload file's type, refer ${@link UploadFileConstant}
-     * @return {@link com.dlink.common.result.Result}
+     * @return {@link Result}
      */
     Result upload(MultipartFile file, String dir, Byte fileType);
 
@@ -63,8 +63,25 @@ public interface FileUploadService {
      * @param files    {@link MultipartFile} instance
      * @param dir      Local absolute dir
      * @param fileType Upload file's type, refer ${@link UploadFileConstant}
-     * @return {@link com.dlink.common.result.Result}
+     * @return {@link Result}
      */
     Result upload(MultipartFile[] files, String dir, Byte fileType);
 
+    /**
+     * Upload one hdfs file, if target file exists, will delete it first
+     *
+     * @param file             {@link MultipartFile} instance
+     * @param hadoopConfigPath core-site.xml,hdfs-site.xml,yarn-site.xml
+     * @return {@link Result}
+     */
+    Result uploadHdfs(MultipartFile file, String dir, String hadoopConfigPath);
+
+    /**
+     * Upload multy hdfs file, if target file exists, will delete it first
+     *
+     * @param files            {@link MultipartFile} instance
+     * @param hadoopConfigPath core-site.xml,hdfs-site.xml,yarn-site.xml
+     * @return {@link Result}
+     */
+    Result uploadHdfs(MultipartFile[] files, String dir, String hadoopConfigPath);
 }

dlink-admin/src/main/java/com/dlink/service/impl/FileUploadServiceImpl.java

@@ -19,6 +19,7 @@
 
 package com.dlink.service.impl;
 
+import com.dlink.assertion.Asserts;
 import com.dlink.common.result.Result;
 import com.dlink.constant.UploadFileConstant;
 import com.dlink.model.CodeEnum;
@@ -48,8 +49,6 @@ import lombok.extern.slf4j.Slf4j;
 @Service
 public class FileUploadServiceImpl implements FileUploadService {
 
-    @Resource
-    private HdfsUtil hdfsUtil;
     @Resource
     private UploadFileRecordService uploadFileRecordService;
 
@@ -76,7 +75,7 @@ public class FileUploadServiceImpl implements FileUploadService {
                 }
             }
             case UploadFileConstant.TARGET_HDFS: {
-                Result result = hdfsUtil.uploadFile(filePath, file);
+                Result result = HdfsUtil.uploadFile(filePath, file);
                 if (Objects.equals(result.getCode(), CodeEnum.SUCCESS.getCode())) {
                     if (uploadFileRecordService.saveOrUpdateFile(file.getOriginalFilename(), dir, filePath, fileType, UploadFileConstant.TARGET_HDFS)) {
                         return Result.succeed("上传成功");
@@ -119,6 +118,42 @@ public class FileUploadServiceImpl implements FileUploadService {
         }
     }
 
+    @Override
+    public Result uploadHdfs(MultipartFile file, String dir, String hadoopConfigPath) {
+        String filePath = FilePathUtil.addFileSeparator(dir) + file.getOriginalFilename();
+        Result result = HdfsUtil.uploadFile(filePath, file, hadoopConfigPath);
+        if (Objects.equals(result.getCode(), CodeEnum.SUCCESS.getCode())) {
+            if (uploadFileRecordService.saveOrUpdateFile(file.getOriginalFilename(), dir, filePath, UploadFileConstant.FLINK_LIB_ID, UploadFileConstant.TARGET_HDFS)) {
+                return Result.succeed("上传成功");
+            } else {
+                return Result.failed("数据库异常");
+            }
+        } else {
+            return result;
+        }
+    }
+
+    @Override
+    public Result uploadHdfs(MultipartFile[] files, String dir, String hadoopConfigPath) {
+        if (Asserts.isNullString(dir)) {
+            dir = UploadFileConstant.getDirPath(UploadFileConstant.FLINK_LIB_ID);
+        }
+        if (files.length > 0) {
+            for (MultipartFile file : files) {
+                Result uploadResult = uploadHdfs(file, dir, hadoopConfigPath);
+                if (Objects.equals(uploadResult.getCode(), CodeEnum.ERROR.getCode())) {
+                    return uploadResult;
+                }
+            }
+            if (!uploadFileRecordService.saveOrUpdateDir(dir, UploadFileConstant.FLINK_LIB_ID, UploadFileConstant.TARGET_HDFS)) {
+                return Result.failed("数据库异常");
+            }
+            return Result.succeed("全部上传成功");
+        } else {
+            return Result.succeed("没有检测到要上传的文件");
+        }
+    }
+
     @Override
     public Result upload(MultipartFile[] files, Byte fileType) {
         String dir = UploadFileConstant.getDirPath(fileType);

dlink-admin/src/main/java/com/dlink/utils/HdfsUtil.java

 package com.dlink.utils;
 
+import com.dlink.assertion.Asserts;
 import com.dlink.common.result.Result;
 import com.dlink.constant.UploadFileConstant;
 import com.dlink.model.CodeEnum;
@@ -13,9 +14,6 @@ import java.io.File;
 import java.io.IOException;
 import java.util.Objects;
 
-import javax.annotation.PostConstruct;
-
-import org.springframework.stereotype.Component;
 import org.springframework.web.multipart.MultipartFile;
 
 import cn.hutool.core.exceptions.ExceptionUtil;
@@ -25,24 +23,27 @@ import lombok.extern.slf4j.Slf4j;
  * Hdfs Handle
  **/
 @Slf4j
-@Component
 public class HdfsUtil {
 
-    private final Configuration configuration = new Configuration();
-    private FileSystem hdfs = null;
+    private static FileSystem hdfs = null;
 
     /**
      * Init internal hdfs client
+     *
+     * @param hadoopConfigPath HDFS config path
      */
-    @PostConstruct
-    private Result init() {
+    private static Result init(String hadoopConfigPath) {
         if (hdfs == null) {
-            String coreSiteFilePath = FilePathUtil.addFileSeparator(UploadFileConstant.HADOOP_CONF_DIR) + "core-site.xml";
-            String hdfsSiteFilePath = FilePathUtil.addFileSeparator(UploadFileConstant.HADOOP_CONF_DIR) + "hdfs-site.xml";
+            if (Asserts.isNullString(hadoopConfigPath)) {
+                hadoopConfigPath = FilePathUtil.removeFileSeparator(UploadFileConstant.HADOOP_CONF_DIR);
+            }
+            String coreSiteFilePath = hadoopConfigPath + "/core-site.xml";
+            String hdfsSiteFilePath = hadoopConfigPath + "/hdfs-site.xml";
             if (!new File(coreSiteFilePath).exists() || !new File(hdfsSiteFilePath).exists()) {
                 return Result.failed("在项目根目录下没有找到 core-site.xml/hdfs-site.xml/yarn-site.xml 文件，请先上传这些文件");
             }
             try {
+                final Configuration configuration = new Configuration();
                 configuration.addResource(new Path(coreSiteFilePath));
                 configuration.addResource(new Path(hdfsSiteFilePath));
                 hdfs = FileSystem.get(configuration);
@@ -60,10 +61,22 @@ public class HdfsUtil {
      *
      * @param path  HDFS path
      * @param bytes File byte content
-     * @return {@link com.dlink.common.result.Result}
+     * @return {@link Result}
+     */
+    public static Result uploadFile(String path, byte[] bytes) {
+        return uploadFile(path, bytes, null);
+    }
+
+    /**
+     * Upload file byte content to HDFS
+     *
+     * @param path             HDFS path
+     * @param bytes            File byte content
+     * @param hadoopConfigPath hdfs config path
+     * @return {@link Result}
      */
-    public Result uploadFile(String path, byte[] bytes) {
-        Result initResult = init();
+    public static Result uploadFile(String path, byte[] bytes, String hadoopConfigPath) {
+        Result initResult = init(hadoopConfigPath);
         if (Objects.equals(initResult.getCode(), CodeEnum.SUCCESS.getCode())) {
             try (FSDataOutputStream stream = hdfs.create(new Path(path), true)) {
                 stream.write(bytes);
@@ -83,9 +96,9 @@ public class HdfsUtil {
      *
      * @param path HDFS path
      * @param file MultipartFile instance
-     * @return {@link com.dlink.common.result.Result}
+     * @return {@link Result}
      */
-    public Result uploadFile(String path, MultipartFile file) {
+    public static Result uploadFile(String path, MultipartFile file) {
         try {
             return uploadFile(path, file.getBytes());
         } catch (IOException e) {
@@ -94,4 +107,21 @@ public class HdfsUtil {
         }
     }
 
+    /**
+     * Upload file byte content to HDFS
+     *
+     * @param path             HDFS path
+     * @param file             MultipartFile instance
+     * @param hadoopConfigPath hdfs config path
+     * @return {@link Result}
+     */
+    public static Result uploadFile(String path, MultipartFile file, String hadoopConfigPath) {
+        try {
+            return uploadFile(path, file.getBytes(), hadoopConfigPath);
+        } catch (IOException e) {
+            log.error(ExceptionUtil.stacktraceToString(e));
+            return Result.failed("文件上传失败");
+        }
+    }
+
 }

dlink-app/dlink-app-base/src/main/java/com/dlink/app/flinksql/Submiter.java

@@ -28,6 +28,7 @@ import com.dlink.executor.ExecutorSetting;
 import com.dlink.interceptor.FlinkInterceptor;
 import com.dlink.parser.SqlType;
 import com.dlink.trans.Operations;
+import com.dlink.utils.SqlUtil;
 
 import org.apache.flink.configuration.CheckpointingOptions;
 
@@ -65,8 +66,8 @@ public class Submiter {
             throw new SQLException("请指定任务ID");
         }
         return "select id, name, alias as jobName, type,check_point as checkpoint,"
-                + "save_point_path as savePointPath, parallelism,fragment as useSqlFragment,statement_set as useStatementSet,config_json as config,"
-                + " env_id as envId,batch_model AS useBatchModel from dlink_task where id = " + id;
+            + "save_point_path as savePointPath, parallelism,fragment as useSqlFragment,statement_set as useStatementSet,config_json as config,"
+            + " env_id as envId,batch_model AS useBatchModel from dlink_task where id = " + id;
     }
 
     private static String getFlinkSQLStatement(Integer id, DBConfig config) {
@@ -90,7 +91,7 @@ public class Submiter {
     }
 
     public static List<String> getStatements(String sql) {
-        return Arrays.asList(sql.split(FlinkSQLConstant.SEPARATOR));
+        return Arrays.asList(SqlUtil.getStatements(sql));
     }
 
     public static void submit(Integer id, DBConfig dbConfig) {
@@ -110,11 +111,11 @@ public class Submiter {
         String uuid = UUID.randomUUID().toString().replace("-", "");
         if (executorSetting.getConfig().containsKey(CheckpointingOptions.CHECKPOINTS_DIRECTORY.key())) {
             executorSetting.getConfig().put(CheckpointingOptions.CHECKPOINTS_DIRECTORY.key(),
-                    executorSetting.getConfig().get(CheckpointingOptions.CHECKPOINTS_DIRECTORY.key()) + "/" + uuid);
+                executorSetting.getConfig().get(CheckpointingOptions.CHECKPOINTS_DIRECTORY.key()) + "/" + uuid);
         }
         if (executorSetting.getConfig().containsKey(CheckpointingOptions.SAVEPOINT_DIRECTORY.key())) {
             executorSetting.getConfig().put(CheckpointingOptions.SAVEPOINT_DIRECTORY.key(),
-                    executorSetting.getConfig().get(CheckpointingOptions.SAVEPOINT_DIRECTORY.key()) + "/" + uuid);
+                executorSetting.getConfig().get(CheckpointingOptions.SAVEPOINT_DIRECTORY.key()) + "/" + uuid);
         }
         logger.info("作业配置如下： {}", executorSetting);
         Executor executor = Executor.buildAppStreamExecutor(executorSetting);

dlink-client/dlink-client-1.11/src/main/java/com/dlink/cdc/AbstractSinkBuilder.java

@@ -193,7 +193,7 @@ public abstract class AbstractSinkBuilder {
                             default:
                         }
                     } catch (Exception e) {
-                        logger.error("SchameTable: {} - Row: {} - Exception: {}", schemaTableName, JSONUtil.toJsonString(value), e.getCause().getMessage());
+                        logger.error("SchameTable: {} - Row: {} - Exception:", schemaTableName, JSONUtil.toJsonString(value), e);
                         throw e;
                     }
                 }

dlink-client/dlink-client-1.11/src/main/java/com/dlink/cdc/sql/SQLSinkBuilder.java

@@ -144,7 +144,7 @@ public class SQLSinkBuilder extends AbstractSinkBuilder implements SinkBuilder,
                                 default:
                             }
                         } catch (Exception e) {
-                            logger.error("SchameTable: {} - Row: {} - Exception: {}", schemaTableName, JSONUtil.toJsonString(value), e.getCause().getMessage());
+                            logger.error("SchameTable: {} - Row: {} - Exception:", schemaTableName, JSONUtil.toJsonString(value), e);
                             throw e;
                         }
                     }

dlink-client/dlink-client-1.12/src/main/java/com/dlink/cdc/AbstractSinkBuilder.java

@@ -193,7 +193,7 @@ public abstract class AbstractSinkBuilder {
                             default:
                         }
                     } catch (Exception e) {
-                        logger.error("SchameTable: {} - Row: {} - Exception: {}", schemaTableName, JSONUtil.toJsonString(value), e.getCause().getMessage());
+                        logger.error("SchameTable: {} - Row: {} - Exception:", schemaTableName, JSONUtil.toJsonString(value), e);
                         throw e;
                     }
                 }

dlink-client/dlink-client-1.12/src/main/java/com/dlink/cdc/sql/SQLSinkBuilder.java

@@ -144,7 +144,7 @@ public class SQLSinkBuilder extends AbstractSinkBuilder implements SinkBuilder,
                                 default:
                             }
                         } catch (Exception e) {
-                            logger.error("SchameTable: {} - Row: {} - Exception: {}", schemaTableName, JSONUtil.toJsonString(value), e.getCause().getMessage());
+                            logger.error("SchameTable: {} - Row: {} - Exception:", schemaTableName, JSONUtil.toJsonString(value), e);
                             throw e;
                         }
                     }

dlink-client/dlink-client-1.13/src/main/java/com/dlink/cdc/AbstractSinkBuilder.java

@@ -193,7 +193,7 @@ public abstract class AbstractSinkBuilder {
                             default:
                         }
                     } catch (Exception e) {
-                        logger.error("SchameTable: {} - Row: {} - Exception: {}", schemaTableName, JSONUtil.toJsonString(value), e.getCause().getMessage());
+                        logger.error("SchameTable: {} - Row: {} - Exception:", schemaTableName, JSONUtil.toJsonString(value), e);
                         throw e;
                     }
                 }

dlink-client/dlink-client-1.13/src/main/java/com/dlink/cdc/sql/SQLSinkBuilder.java

@@ -144,7 +144,7 @@ public class SQLSinkBuilder extends AbstractSinkBuilder implements SinkBuilder,
                                 default:
                             }
                         } catch (Exception e) {
-                            logger.error("SchameTable: {} - Row: {} - Exception: {}", schemaTableName, JSONUtil.toJsonString(value), e.getCause().getMessage());
+                            logger.error("SchameTable: {} - Row: {} - Exception:", schemaTableName, JSONUtil.toJsonString(value), e);
                             throw e;
                         }
                     }

dlink-client/dlink-client-1.14/src/main/java/com/dlink/cdc/AbstractSinkBuilder.java

@@ -59,7 +59,6 @@ import org.apache.flink.util.Collector;
 import org.apache.flink.util.OutputTag;
 
 import java.math.BigDecimal;
-import java.sql.Timestamp;
 import java.time.Instant;
 import java.time.ZoneId;
 import java.util.ArrayList;
@@ -68,6 +67,8 @@ import java.util.List;
 import java.util.Map;
 import java.util.Properties;
 
+import javax.xml.bind.DatatypeConverter;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -83,6 +84,7 @@ public abstract class AbstractSinkBuilder {
 
     protected FlinkCDCConfig config;
     protected List<ModifyOperation> modifyOperations = new ArrayList();
+    private ZoneId sinkTimeZone = ZoneId.of("UTC");
 
     public AbstractSinkBuilder() {
     }
@@ -104,7 +106,7 @@ public abstract class AbstractSinkBuilder {
         Map<String, String> sink = config.getSink();
         for (Map.Entry<String, String> entry : sink.entrySet()) {
             if (Asserts.isNotNullString(entry.getKey()) && entry.getKey().startsWith("sink.properties") && Asserts.isNotNullString(entry.getValue())) {
-                properties.setProperty(entry.getKey().replace("sink.properties.",""), entry.getValue());
+                properties.setProperty(entry.getKey().replace("sink.properties.", ""), entry.getValue());
             }
         }
         return properties;
@@ -137,9 +139,9 @@ public abstract class AbstractSinkBuilder {
     }
 
     protected DataStream<Map> shunt(
-            SingleOutputStreamOperator<Map> processOperator,
-            Table table,
-            OutputTag<Map> tag) {
+        SingleOutputStreamOperator<Map> processOperator,
+        Table table,
+        OutputTag<Map> tag) {
 
         return processOperator.getSideOutput(tag);
     }
@@ -193,7 +195,7 @@ public abstract class AbstractSinkBuilder {
                             default:
                         }
                     } catch (Exception e) {
-                        logger.error("SchameTable: {} - Row: {} - Exception: {}", schemaTableName, JSONUtil.toJsonString(value), e.getCause().getMessage());
+                        logger.error("SchameTable: {} - Row: {} - Exception: {}", schemaTableName, JSONUtil.toJsonString(value), e);
                         throw e;
                     }
                 }
@@ -213,6 +215,12 @@ public abstract class AbstractSinkBuilder {
         CustomTableEnvironment customTableEnvironment,
         DataStreamSource<String> dataStreamSource) {
 
+        final String timeZone = config.getSink().get("timezone");
+        config.getSink().remove("timezone");
+        if (Asserts.isNotNullString(timeZone)) {
+            sinkTimeZone = ZoneId.of(timeZone);
+        }
+
         final List<Schema> schemaList = config.getSchemaList();
         final String schemaFieldName = config.getSchemaFieldName();
 
@@ -294,14 +302,41 @@ public abstract class AbstractSinkBuilder {
         if (logicalType instanceof VarCharType) {
             return StringData.fromString((String) value);
         } else if (logicalType instanceof DateType) {
-            return StringData.fromString(Instant.ofEpochMilli((long) value).atZone(ZoneId.systemDefault()).toLocalDate().toString());
+            return value;
         } else if (logicalType instanceof TimestampType) {
-            return TimestampData.fromTimestamp(Timestamp.from(Instant.ofEpochMilli((long) value)));
+            if (value instanceof Integer) {
+                return TimestampData.fromLocalDateTime(Instant.ofEpochMilli(((Integer) value).longValue()).atZone(sinkTimeZone).toLocalDateTime());
+            } else if (value instanceof Long) {
+                return TimestampData.fromLocalDateTime(Instant.ofEpochMilli((long) value).atZone(sinkTimeZone).toLocalDateTime());
+            } else {
+                return TimestampData.fromLocalDateTime(Instant.parse(value.toString()).atZone(sinkTimeZone).toLocalDateTime());
+            }
         } else if (logicalType instanceof DecimalType) {
             final DecimalType decimalType = ((DecimalType) logicalType);
             final int precision = decimalType.getPrecision();
             final int scale = decimalType.getScale();
             return DecimalData.fromBigDecimal(new BigDecimal((String) value), precision, scale);
+        } else if (logicalType instanceof FloatType) {
+            if (value instanceof Float) {
+                return value;
+            } else if (value instanceof Double) {
+                return ((Double) value).floatValue();
+            } else {
+                return Float.parseFloat(value.toString());
+            }
+        } else if (logicalType instanceof BigIntType) {
+            if (value instanceof Integer) {
+                return ((Integer) value).longValue();
+            } else {
+                return value;
+            }
+        } else if (logicalType instanceof VarBinaryType) {
+            // VARBINARY AND BINARY is converted to String with encoding base64 in FlinkCDC.
+            if (value instanceof String) {
+                return DatatypeConverter.parseBase64Binary(value.toString());
+            } else {
+                return value;
+            }
         } else {
             return value;
         }

dlink-client/dlink-client-1.14/src/main/java/com/dlink/cdc/doris/DorisSinkBuilder.java

@@ -40,6 +40,7 @@ import java.io.Serializable;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.Properties;
 
 /**
  * DorisSinkBuilder
@@ -137,6 +138,8 @@ public class DorisSinkBuilder extends AbstractSinkBuilder implements SinkBuilder
         }
         if (sink.containsKey(DorisSinkOptions.SINK_ENABLE_DELETE.key())) {
             executionBuilder.setDeletable(Boolean.valueOf(sink.get(DorisSinkOptions.SINK_ENABLE_DELETE.key())));
+        } else {
+            executionBuilder.setDeletable(true);
         }
         if (sink.containsKey(DorisSinkOptions.SINK_LABEL_PREFIX.key())) {
             executionBuilder.setLabelPrefix(getSinkSchemaName(table) + "_" + getSinkTableName(table) + sink.get(DorisSinkOptions.SINK_LABEL_PREFIX.key()));
@@ -144,7 +147,12 @@ public class DorisSinkBuilder extends AbstractSinkBuilder implements SinkBuilder
         if (sink.containsKey(DorisSinkOptions.SINK_MAX_RETRIES.key())) {
             executionBuilder.setMaxRetries(Integer.valueOf(sink.get(DorisSinkOptions.SINK_MAX_RETRIES.key())));
         }
-        executionBuilder.setStreamLoadProp(getProperties());
+
+        Properties properties = getProperties();
+        // Doris 1.1 need to this para to support delete
+        properties.setProperty("columns", String.join(",", columnNameList) + ",__DORIS_DELETE_SIGN__");
+
+        executionBuilder.setStreamLoadProp(properties);
 
         // Create DorisSink.
         DorisSink.Builder<RowData> builder = DorisSink.builder();
@@ -153,6 +161,7 @@ public class DorisSinkBuilder extends AbstractSinkBuilder implements SinkBuilder
             .setSerializer(RowDataSerializer.builder()
                 .setFieldNames(columnNames)
                 .setType("json")
+                .enableDelete(true)
                 .setFieldType(columnTypes).build())
             .setDorisOptions(dorisBuilder.build());
 

dlink-client/dlink-client-1.14/src/main/java/com/dlink/cdc/sql/SQLSinkBuilder.java

@@ -145,7 +145,7 @@ public class SQLSinkBuilder extends AbstractSinkBuilder implements SinkBuilder,
                             default:
                         }
                     } catch (Exception e) {
-                        logger.error("SchameTable: {} - Row: {} - Exception: {}", schemaTableName, JSONUtil.toJsonString(value), e.getCause().getMessage());
+                        logger.error("SchameTable: {} - Row: {} - Exception:", schemaTableName, JSONUtil.toJsonString(value),e);
                         throw e;
                     }
                 }

dlink-client/dlink-client-1.15/src/main/java/com/dlink/cdc/AbstractSinkBuilder.java

@@ -193,7 +193,7 @@ public abstract class AbstractSinkBuilder {
                             default:
                         }
                     } catch (Exception e) {
-                        logger.error("SchameTable: {} - Row: {} - Exception: {}", schemaTableName, JSONUtil.toJsonString(value), e.getCause().getMessage());
+                        logger.error("SchameTable: {} - Row: {} - Exception:", schemaTableName, JSONUtil.toJsonString(value), e);
                         throw e;
                     }
                 }

dlink-client/dlink-client-1.15/src/main/java/com/dlink/cdc/sql/SQLSinkBuilder.java

@@ -144,7 +144,7 @@ public class SQLSinkBuilder extends AbstractSinkBuilder implements SinkBuilder,
                                 default:
                             }
                         } catch (Exception e) {
-                            logger.error("SchameTable: {} - Row: {} - Exception: {}", schemaTableName, JSONUtil.toJsonString(value), e.getCause().getMessage());
+                            logger.error("SchameTable: {} - Row: {} - Exception:", schemaTableName, JSONUtil.toJsonString(value), e);
                             throw e;
                         }
                     }

dlink-core/src/main/java/com/dlink/api/FlinkAPI.java

@@ -207,82 +207,82 @@ public class FlinkAPI {
     }
 
     /**
+     * @return JsonNode
      * @Author: zhumingye
      * @date: 2022/6/24
      * @Description: getJobManagerMetrics 获取jobManager的监控信息
-     * @return JsonNode
      */
     public JsonNode getJobManagerMetrics() {
         return get(FlinkRestAPIConstant.JOB_MANAGER + FlinkRestAPIConstant.METRICS + FlinkRestAPIConstant.GET + buildMetricsParms(FlinkRestAPIConstant.JOB_MANAGER));
     }
 
     /**
+     * @return JsonNode
      * @Author: zhumingye
      * @date: 2022/6/24
      * @Description: getJobManagerConfig 获取jobManager的配置信息
-     * @return JsonNode
      */
     public JsonNode getJobManagerConfig() {
         return get(FlinkRestAPIConstant.JOB_MANAGER + FlinkRestAPIConstant.CONFIG);
     }
 
     /**
+     * @return JsonNode
      * @Author: zhumingye
      * @date: 2022/6/24
      * @Description: getJobManagerLog 获取jobManager的日志信息
-     * @return JsonNode
      */
     public String getJobManagerLog() {
         return getResult(FlinkRestAPIConstant.JOB_MANAGER + FlinkRestAPIConstant.LOG);
     }
 
     /**
+     * @return String
      * @Author: zhumingye
      * @date: 2022/6/24
      * @Description: getJobManagerStdOut 获取jobManager的控制台输出日志
-     * @return String
      */
     public String getJobManagerStdOut() {
         return getResult(FlinkRestAPIConstant.JOB_MANAGER + FlinkRestAPIConstant.STDOUT);
     }
 
     /**
+     * @return JsonNode
      * @Author: zhumingye
      * @date: 2022/6/24
      * @Description: getJobManagerLogList 获取jobManager的日志列表
-     * @return JsonNode
      */
     public JsonNode getJobManagerLogList() {
         return get(FlinkRestAPIConstant.JOB_MANAGER + FlinkRestAPIConstant.LOGS);
     }
 
     /**
+     * @param logName 日志文件名
+     * @return String
      * @Author: zhumingye
      * @date: 2022/6/24
      * @Description: getJobManagerLogFileDetail 获取jobManager的日志文件的具体信息
-     * @param logName 日志文件名
-     * @return String
      */
     public String getJobManagerLogFileDetail(String logName) {
         return getResult(FlinkRestAPIConstant.JOB_MANAGER + FlinkRestAPIConstant.LOGS + logName);
     }
 
     /**
+     * @return JsonNode
      * @Author: zhumingye
      * @date: 2022/6/24
      * @Description: getTaskManagers 获取taskManager的列表
-     * @return JsonNode
      */
     public JsonNode getTaskManagers() {
         return get(FlinkRestAPIConstant.TASK_MANAGER);
     }
 
     /**
+     * @return String
      * @Author: zhumingye
      * @date: 2022/6/24
      * @Description: buildMetricsParms 构建metrics参数
      * @Params: type:  入参类型 可选值：task-manager, job-manager
-     * @return String
      */
     public String buildMetricsParms(String type) {
         JsonNode jsonNode = get(type + FlinkRestAPIConstant.METRICS);
@@ -290,70 +290,75 @@ public class FlinkAPI {
         Iterator<JsonNode> jsonNodeIterator = jsonNode.elements();
         while (jsonNodeIterator.hasNext()) {
             JsonNode node = jsonNodeIterator.next();
-            sb.append(node.get("id").asText()).append(",");
+            if (Asserts.isNotNull(node) && Asserts.isNotNull(node.get("id"))) {
+                if (sb.length() > 0) {
+                    sb.append(",");
+                }
+                sb.append(node.get("id").asText());
+            }
         }
-        return sb.deleteCharAt(sb.length() - 1).toString();
+        return sb.toString();
     }
 
     /**
+     * @return JsonNode
      * @Author: zhumingye
      * @date: 2022/6/24
      * @Description: getJobManagerLog 获取jobManager的日志信息
-     * @return JsonNode
      */
     public JsonNode getTaskManagerMetrics(String containerId) {
         return get(FlinkRestAPIConstant.TASK_MANAGER + containerId + FlinkRestAPIConstant.METRICS + FlinkRestAPIConstant.GET + buildMetricsParms(FlinkRestAPIConstant.JOB_MANAGER));
     }
 
     /**
+     * @param containerId 容器id
+     * @return String
      * @Author: zhumingye
      * @date: 2022/6/24
      * @Description: getTaskManagerLog 获取taskManager的日志信息
-     * @param containerId 容器id
-     * @return String
      */
     public String getTaskManagerLog(String containerId) {
         return getResult(FlinkRestAPIConstant.TASK_MANAGER + containerId + FlinkRestAPIConstant.LOG);
     }
 
     /**
+     * @param containerId 容器id
+     * @return JsonNode
      * @Author: zhumingye
      * @date: 2022/6/24
      * @Description: getTaskManagerStdOut 获取taskManager的StdOut日志信息
-     * @param containerId 容器id
-     * @return JsonNode
      */
     public String getTaskManagerStdOut(String containerId) {
         return getResult(FlinkRestAPIConstant.TASK_MANAGER + containerId + FlinkRestAPIConstant.STDOUT);
     }
 
     /**
+     * @param containerId 容器id
+     * @return JsonNode
      * @Author: zhumingye
      * @date: 2022/6/24
      * @Description: getTaskManagerLogList 获取taskManager的日志列表
-     * @param containerId 容器id
-     * @return JsonNode
      */
     public JsonNode getTaskManagerLogList(String containerId) {
         return get(FlinkRestAPIConstant.TASK_MANAGER + containerId + FlinkRestAPIConstant.LOGS);
     }
 
     /**
+     * @param logName 日志名称
+     * @return String
      * @Author: zhumingye
      * @date: 2022/6/24
      * @Description: getTaskManagerLogFileDeatil 获取具体日志的详细信息
-     * @param logName 日志名称
-     * @return String
      */
-    public String getTaskManagerLogFileDeatil(String containerId,String logName) {
+    public String getTaskManagerLogFileDeatil(String containerId, String logName) {
         return getResult(FlinkRestAPIConstant.TASK_MANAGER + containerId + FlinkRestAPIConstant.LOGS + logName);
     }
 
     /**
+     * @return JsonNode
      * @Author: zhumingye
      * @date: 2022/6/24
      * @Description: getTaskManagerThreadDump 获取taskManager的线程信息
-     * @return JsonNode
      */
     public JsonNode getTaskManagerThreadDump(String containerId) {
         return get(FlinkRestAPIConstant.TASK_MANAGER + containerId + FlinkRestAPIConstant.THREAD_DUMP);

dlink-flink/dlink-flink-1.15/pom.xml

@@ -31,7 +31,7 @@
     <properties>
         <java.version>1.8</java.version>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-        <flink.version>1.15.1</flink.version>
+        <flink.version>1.15.2</flink.version>
         <flink.guava.version>15.0</flink.guava.version>
         <flinkcdc.version>2.2.1</flinkcdc.version>
         <commons.version>1.3.1</commons.version>

dlink-web/src/global.less

@@ -82,11 +82,11 @@ ol {
   margin-bottom: 0!important;
 }
 
-div .ant-pro-card-body {
-  padding: 2px;
-}
+//div .ant-pro-card-body {
+//  padding: 2px;
+//}
 div .ant-pro-page-container{
-  padding-top: 1px;
+  padding-top: 0px;
 }
 
 div .ant-page-header {
@@ -98,5 +98,5 @@ div .ant-pro-page-container-children-content{
 }
 
 .ant-tabs-top > .ant-tabs-nav, .ant-tabs-bottom > .ant-tabs-nav, .ant-tabs-top > div > .ant-tabs-nav, .ant-tabs-bottom > div > .ant-tabs-nav {
-  margin: 0 0 4px 0!important;
+  margin: 0 0 0px 0!important;
 }

dlink-web/src/pages/ClusterConfiguration/components/ClusterConfigurationForm.tsx

@@ -102,7 +102,6 @@ const ClusterConfigurationForm: React.FC<ClusterConfigurationFormProps> = (props
     return {
       name: 'files',
       action: '/api/fileUpload',
-      // accept: 'application/json',
       headers: {
         authorization: 'authorization-text',
       },
@@ -124,6 +123,32 @@ const ClusterConfigurationForm: React.FC<ClusterConfigurationFormProps> = (props
     }
   };
 
+  const getUploadHdfsProps = (dir: string) => {
+    return {
+      name: 'files',
+      action: '/api/fileUpload/hdfs',
+      headers: {
+        authorization: 'authorization-text',
+      },
+      data: {
+        dir,
+        hadoopConfigPath
+      },
+      showUploadList: true,
+      onChange(info) {
+        if (info.file.status === 'done') {
+          if (info.file.response.code == CODE.SUCCESS) {
+            message.success(info.file.response.msg);
+          } else {
+            message.warn(info.file.response.msg);
+          }
+        } else if (info.file.status === 'error') {
+          message.error(`${info.file.name} 上传失败`);
+        }
+      },
+    }
+  };
+
   const renderContent = (formValsPara: Partial<ClusterConfigurationTableListItem>) => {
     return (
       <>
@@ -146,7 +171,7 @@ const ClusterConfigurationForm: React.FC<ClusterConfigurationFormProps> = (props
           >
             <Input placeholder="值如 /etc/hadoop/conf" addonAfter={
               <Form.Item name="suffix" noStyle>
-                <Upload {...getUploadProps(hadoopConfigPath)}>
+                <Upload {...getUploadProps(hadoopConfigPath)} multiple>
                   <UploadOutlined/>
                 </Upload>
               </Form.Item>}/>
@@ -235,7 +260,7 @@ const ClusterConfigurationForm: React.FC<ClusterConfigurationFormProps> = (props
           >
             <Input placeholder="值如 hdfs:///flink/lib" addonAfter={
               <Form.Item name="suffix" noStyle>
-                <Upload {...getUploadProps(flinkLibPath)}>
+                <Upload {...getUploadHdfsProps(flinkLibPath)} multiple>
                   <UploadOutlined/>
                 </Upload>
               </Form.Item>}/>

docs/docs/quick_start/dbql_quick_start.md

@@ -10,7 +10,7 @@ title: DB SQL 作业快速入门
 
 ### 创建数据源
 
-选择**注册中心>>数据源管理>>新建**，假设您连接Starrocks。
+选择**注册中心>>数据源管理>>新建**，假设您连接Doris。
 
 ![createSource](http://www.aiwenmo.com/dinky/docs/zh-CN/quick_start/dbql_quick_start/createSource.png)