[Fix][core,client] Resolve the exception when starting from the specified savepoint (#1041)

* fix ：解决从指定的savepoint恢复任务时，未设置savepint文件路径，以及从savepoint启动未匹配到算子状态跳过启动

* 增加mysql cdc的参数配置

* 优化datastream kafka-json 和datastream starrocks,能被合并进去

* 修复sinkbuilder返回对象和方法重命名

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

@@ -757,6 +757,7 @@ public class TaskServiceImpl extends SuperServiceImpl<TaskMapper, Task> implemen
                 }
                 break;
             case CUSTOM:
+                config.setSavePointPath(config.getSavePointPath());
                 config.getConfig().put("execution.savepoint.path", config.getSavePointPath());
                 break;
             default:

dlink-client/dlink-client-1.13/src/main/java/com/dlink/cdc/mysql/MysqlCDCBuilder.java

@@ -80,6 +80,10 @@ public class MysqlCDCBuilder extends AbstractCDCBuilder implements CDCBuilder {
         String connectMaxRetries = config.getSource().get("connect.max-retries");
         String connectionPoolSize = config.getSource().get("connection.pool.size");
         String heartbeatInterval = config.getSource().get("heartbeat.interval");
+        String chunkSize = config.getSource().get("scan.incremental.snapshot.chunk.size");
+        String distributionFactorLower = config.getSource().get("chunk-key.even-distribution.factor.upper-bound");
+        String distributionFactorUpper = config.getSource().get("chunk-key.even-distribution.factor.lower-bound");
+        String scanNewlyAddedTableEnabled = config.getSource().get("scan.newly-added-table.enabled");
 
         Properties debeziumProperties = new Properties();
         // 为部分转换添加默认值
@@ -123,7 +127,9 @@ public class MysqlCDCBuilder extends AbstractCDCBuilder implements CDCBuilder {
         sourceBuilder.deserializer(new MysqlJsonDebeziumDeserializationSchema());
         sourceBuilder.debeziumProperties(debeziumProperties);
         sourceBuilder.jdbcProperties(jdbcProperties);
-
+        if ("true".equalsIgnoreCase(scanNewlyAddedTableEnabled)) {
+            sourceBuilder.scanNewlyAddedTableEnabled(true);
+        }
         if (Asserts.isNotNullString(config.getStartupMode())) {
             switch (config.getStartupMode().toLowerCase()) {
                 case "initial":
@@ -165,7 +171,15 @@ public class MysqlCDCBuilder extends AbstractCDCBuilder implements CDCBuilder {
         if (Asserts.isNotNullString(heartbeatInterval)) {
             sourceBuilder.heartbeatInterval(Duration.ofMillis(Long.valueOf(heartbeatInterval)));
         }
-
+        if (Asserts.isAllNotNullString(chunkSize)) {
+            sourceBuilder.splitSize(Integer.parseInt(chunkSize));
+        }
+        if (Asserts.isNotNullString(distributionFactorLower)) {
+            sourceBuilder.distributionFactorLower(Double.valueOf(distributionFactorLower));
+        }
+        if (Asserts.isNotNullString(distributionFactorUpper)) {
+            sourceBuilder.distributionFactorUpper(Double.valueOf(distributionFactorUpper));
+        }
         return env.fromSource(sourceBuilder.build(), WatermarkStrategy.noWatermarks(), "MySQL CDC Source");
     }
 

dlink-core/src/main/java/com/dlink/job/JobManager.java

@@ -376,7 +376,7 @@ public class JobManager {
                         streamGraph.setJobName(config.getJobName());
                         JobGraph jobGraph = streamGraph.getJobGraph();
                         if (Asserts.isNotNullString(config.getSavePointPath())) {
-                            jobGraph.setSavepointRestoreSettings(SavepointRestoreSettings.forPath(config.getSavePointPath()));
+                            jobGraph.setSavepointRestoreSettings(SavepointRestoreSettings.forPath(config.getSavePointPath(),true));
                         }
                         gatewayResult = Gateway.build(config.getGatewayConfig()).submitJobGraph(jobGraph);
                     }
@@ -434,7 +434,7 @@ public class JobManager {
             JobGraph jobGraph = executor.getJobGraphFromInserts(inserts);
             // Perjob mode need to set savepoint restore path, when recovery from savepoint.
             if (Asserts.isNotNullString(config.getSavePointPath())) {
-                jobGraph.setSavepointRestoreSettings(SavepointRestoreSettings.forPath(config.getSavePointPath()));
+                jobGraph.setSavepointRestoreSettings(SavepointRestoreSettings.forPath(config.getSavePointPath(),true));
             }
             // Perjob mode need to submit job graph.
             gatewayResult = Gateway.build(config.getGatewayConfig()).submitJobGraph(jobGraph);