Merge branch 'DataLinkDC:dev' into dev

gaoyan1998 · Oct 27, 2023 · c2ab7ef · c2ab7ef
2 parents bb12e0c + 49aca5b
commit c2ab7ef
Show file tree

Hide file tree

Showing 20 changed files with 486 additions and 587 deletions.
diff --git a/dinky-admin/src/main/java/org/dinky/service/impl/TaskServiceImpl.java b/dinky-admin/src/main/java/org/dinky/service/impl/TaskServiceImpl.java
@@ -23,6 +23,7 @@
 import org.dinky.config.Dialect;
 import org.dinky.context.TenantContextHolder;
 import org.dinky.data.annotations.ProcessStep;
+import org.dinky.data.app.AppParamConfig;
 import org.dinky.data.constant.CommonConstant;
 import org.dinky.data.dto.AbstractStatementDTO;
 import org.dinky.data.dto.DebugDTO;
@@ -95,6 +96,7 @@
 import java.io.Reader;
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
+import java.util.Base64;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -153,15 +155,17 @@ public class TaskServiceImpl extends SuperServiceImpl<TaskMapper, Task> implemen
     private CatalogueService catalogueService;
 
     private String[] buildParams(int id) {
-        return String.format(
-                        "--id %d --driver %s --url %s --username %s --password %s --dinkyAddr %s",
-                        id,
-                        dsProperties.getDriverClassName(),
-                        dsProperties.getUrl(),
-                        dsProperties.getUsername(),
-                        dsProperties.getPassword(),
-                        SystemConfiguration.getInstances().getDinkyAddr())
-                .split(" ");
+        AppParamConfig appParamConfig = AppParamConfig.builder()
+                .taskId(id)
+                .url(dsProperties.getUrl())
+                .username(dsProperties.getUsername())
+                .password(dsProperties.getPassword())
+                .dinkyAddr(SystemConfiguration.getInstances().getDinkyAddr().getValue())
+                .split(SystemConfiguration.getInstances().getSqlSeparator())
+                .build();
+        String encodeParam = Base64.getEncoder()
+                .encodeToString(JsonUtils.toJsonString(appParamConfig).getBytes());
+        return StrFormatter.format("--config {}", encodeParam).split(" ");
     }
 
     @ProcessStep(type = ProcessStepType.SUBMIT_PRECHECK)
@@ -217,6 +221,7 @@ public JobConfig buildJobConfig(TaskDTO task) {
             FlinkClusterConfig flinkClusterCfg =
                     clusterCfgService.getFlinkClusterCfg(config.getClusterConfigurationId());
             flinkClusterCfg.getAppConfig().setUserJarParas(buildParams(config.getTaskId()));
+            flinkClusterCfg.getAppConfig().setUserJarMainAppClass(CommonConstant.DINKY_APP_MAIN_CLASS);
             config.buildGatewayConfig(flinkClusterCfg);
         } else {
             log.info("Init remote cluster");

diff --git a/dinky-app/dinky-app-1.13/src/main/java/org/dinky/app/MainApp.java b/dinky-app/dinky-app-1.13/src/main/java/org/dinky/app/MainApp.java
@@ -19,27 +19,42 @@
 
 package org.dinky.app;
 
-import org.dinky.app.db.DBConfig;
+import org.dinky.app.constant.AppParamConstant;
+import org.dinky.app.db.DBUtil;
 import org.dinky.app.flinksql.Submitter;
-import org.dinky.assertion.Asserts;
-import org.dinky.constant.FlinkParamConstant;
-import org.dinky.utils.FlinkBaseUtil;
+import org.dinky.data.app.AppParamConfig;
+import org.dinky.utils.JsonUtils;
 
-import java.io.IOException;
-import java.util.Map;
+import org.apache.flink.api.java.utils.ParameterTool;
+
+import java.util.Base64;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * MainApp
  *
- * @since 2021/10/27
+ * @since 2022/11/05
  */
 public class MainApp {
 
-    public static void main(String[] args) throws IOException {
-        Map<String, String> params = FlinkBaseUtil.getParamsFromArgs(args);
-        String id = params.get(FlinkParamConstant.ID);
-        Asserts.checkNullString(id, "请配置入参 id ");
-        DBConfig dbConfig = DBConfig.build(params);
-        Submitter.submit(Integer.valueOf(id), dbConfig, params.get(FlinkParamConstant.DINKY_ADDR));
+    private static final Logger log = LoggerFactory.getLogger(Submitter.class);
+
+    public static void main(String[] args) throws Exception {
+        log.info("=========================Start run dinky app job===============================");
+        ParameterTool parameters = ParameterTool.fromArgs(args);
+        boolean isEncrypt = parameters.getBoolean(AppParamConstant.isEncrypt, true);
+        String config = parameters.get(AppParamConstant.config);
+        config = isEncrypt ? new String(Base64.getDecoder().decode(config)) : config;
+        AppParamConfig appConfig = JsonUtils.toJavaBean(config, AppParamConfig.class);
+        try {
+            log.info("dinky app is Ready to run, config is {}", appConfig);
+            DBUtil.init(appConfig);
+            Submitter.submit(appConfig);
+        } catch (Exception e) {
+            log.error("exectue app failed with config: {}", appConfig);
+            throw e;
+        }
     }
 }
diff --git a/dinky-app/dinky-app-1.14/src/main/java/org/dinky/app/MainApp.java b/dinky-app/dinky-app-1.14/src/main/java/org/dinky/app/MainApp.java
@@ -19,27 +19,42 @@
 
 package org.dinky.app;
 
-import org.dinky.app.db.DBConfig;
+import org.dinky.app.constant.AppParamConstant;
+import org.dinky.app.db.DBUtil;
 import org.dinky.app.flinksql.Submitter;
-import org.dinky.assertion.Asserts;
-import org.dinky.constant.FlinkParamConstant;
-import org.dinky.utils.FlinkBaseUtil;
+import org.dinky.data.app.AppParamConfig;
+import org.dinky.utils.JsonUtils;
 
-import java.io.IOException;
-import java.util.Map;
+import org.apache.flink.api.java.utils.ParameterTool;
+
+import java.util.Base64;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * MainApp
  *
- * @since 2021/10/27
+ * @since 2022/11/05
  */
 public class MainApp {
 
-    public static void main(String[] args) throws IOException {
-        Map<String, String> params = FlinkBaseUtil.getParamsFromArgs(args);
-        String id = params.get(FlinkParamConstant.ID);
-        Asserts.checkNullString(id, "请配置入参 id ");
-        DBConfig dbConfig = DBConfig.build(params);
-        Submitter.submit(Integer.valueOf(id), dbConfig, params.get(FlinkParamConstant.DINKY_ADDR));
+    private static final Logger log = LoggerFactory.getLogger(Submitter.class);
+
+    public static void main(String[] args) throws Exception {
+        log.info("=========================Start run dinky app job===============================");
+        ParameterTool parameters = ParameterTool.fromArgs(args);
+        boolean isEncrypt = parameters.getBoolean(AppParamConstant.isEncrypt, true);
+        String config = parameters.get(AppParamConstant.config);
+        config = isEncrypt ? new String(Base64.getDecoder().decode(config)) : config;
+        AppParamConfig appConfig = JsonUtils.toJavaBean(config, AppParamConfig.class);
+        try {
+            log.info("dinky app is Ready to run, config is {}", appConfig);
+            DBUtil.init(appConfig);
+            Submitter.submit(appConfig);
+        } catch (Exception e) {
+            log.error("exectue app failed with config: {}", appConfig);
+            throw e;
+        }
     }
 }
diff --git a/dinky-app/dinky-app-1.15/src/main/java/org/dinky/app/MainApp.java b/dinky-app/dinky-app-1.15/src/main/java/org/dinky/app/MainApp.java
@@ -19,27 +19,42 @@
 
 package org.dinky.app;
 
-import org.dinky.app.db.DBConfig;
+import org.dinky.app.constant.AppParamConstant;
+import org.dinky.app.db.DBUtil;
 import org.dinky.app.flinksql.Submitter;
-import org.dinky.assertion.Asserts;
-import org.dinky.constant.FlinkParamConstant;
-import org.dinky.utils.FlinkBaseUtil;
+import org.dinky.data.app.AppParamConfig;
+import org.dinky.utils.JsonUtils;
 
-import java.io.IOException;
-import java.util.Map;
+import org.apache.flink.api.java.utils.ParameterTool;
+
+import java.util.Base64;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * MainApp
  *
- * @since 2021/10/27
+ * @since 2022/11/05
  */
 public class MainApp {
 
-    public static void main(String[] args) throws IOException {
-        Map<String, String> params = FlinkBaseUtil.getParamsFromArgs(args);
-        String id = params.get(FlinkParamConstant.ID);
-        Asserts.checkNullString(id, "请配置入参 id ");
-        DBConfig dbConfig = DBConfig.build(params);
-        Submitter.submit(Integer.valueOf(id), dbConfig, params.get(FlinkParamConstant.DINKY_ADDR));
+    private static final Logger log = LoggerFactory.getLogger(Submitter.class);
+
+    public static void main(String[] args) throws Exception {
+        log.info("=========================Start run dinky app job===============================");
+        ParameterTool parameters = ParameterTool.fromArgs(args);
+        boolean isEncrypt = parameters.getBoolean(AppParamConstant.isEncrypt, true);
+        String config = parameters.get(AppParamConstant.config);
+        config = isEncrypt ? new String(Base64.getDecoder().decode(config)) : config;
+        AppParamConfig appConfig = JsonUtils.toJavaBean(config, AppParamConfig.class);
+        try {
+            log.info("dinky app is Ready to run, config is {}", appConfig);
+            DBUtil.init(appConfig);
+            Submitter.submit(appConfig);
+        } catch (Exception e) {
+            log.error("exectue app failed with config: {}", appConfig);
+            throw e;
+        }
     }
 }
diff --git a/dinky-app/dinky-app-1.16/src/main/java/org/dinky/app/MainApp.java b/dinky-app/dinky-app-1.16/src/main/java/org/dinky/app/MainApp.java
@@ -19,14 +19,18 @@
 
 package org.dinky.app;
 
-import org.dinky.app.db.DBConfig;
+import org.dinky.app.constant.AppParamConstant;
+import org.dinky.app.db.DBUtil;
 import org.dinky.app.flinksql.Submitter;
-import org.dinky.assertion.Asserts;
-import org.dinky.constant.FlinkParamConstant;
-import org.dinky.utils.FlinkBaseUtil;
+import org.dinky.data.app.AppParamConfig;
+import org.dinky.utils.JsonUtils;
 
-import java.io.IOException;
-import java.util.Map;
+import org.apache.flink.api.java.utils.ParameterTool;
+
+import java.util.Base64;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * MainApp
@@ -35,11 +39,22 @@
  */
 public class MainApp {
 
-    public static void main(String[] args) throws IOException {
-        Map<String, String> params = FlinkBaseUtil.getParamsFromArgs(args);
-        String id = params.get(FlinkParamConstant.ID);
-        Asserts.checkNullString(id, "请配置入参 id ");
-        DBConfig dbConfig = DBConfig.build(params);
-        Submitter.submit(Integer.valueOf(id), dbConfig, params.get(FlinkParamConstant.DINKY_ADDR));
+    private static final Logger log = LoggerFactory.getLogger(Submitter.class);
+
+    public static void main(String[] args) throws Exception {
+        log.info("=========================Start run dinky app job===============================");
+        ParameterTool parameters = ParameterTool.fromArgs(args);
+        boolean isEncrypt = parameters.getBoolean(AppParamConstant.isEncrypt, true);
+        String config = parameters.get(AppParamConstant.config);
+        config = isEncrypt ? new String(Base64.getDecoder().decode(config)) : config;
+        AppParamConfig appConfig = JsonUtils.toJavaBean(config, AppParamConfig.class);
+        try {
+            log.info("dinky app is Ready to run, config is {}", appConfig);
+            DBUtil.init(appConfig);
+            Submitter.submit(appConfig);
+        } catch (Exception e) {
+            log.error("exectue app failed with config: {}", appConfig);
+            throw e;
+        }
     }
 }
diff --git a/dinky-app/dinky-app-1.17/src/main/java/org/dinky/app/MainApp.java b/dinky-app/dinky-app-1.17/src/main/java/org/dinky/app/MainApp.java
@@ -19,14 +19,18 @@
 
 package org.dinky.app;
 
-import org.dinky.app.db.DBConfig;
+import org.dinky.app.constant.AppParamConstant;
+import org.dinky.app.db.DBUtil;
 import org.dinky.app.flinksql.Submitter;
-import org.dinky.assertion.Asserts;
-import org.dinky.constant.FlinkParamConstant;
-import org.dinky.utils.FlinkBaseUtil;
+import org.dinky.data.app.AppParamConfig;
+import org.dinky.utils.JsonUtils;
 
-import java.io.IOException;
-import java.util.Map;
+import org.apache.flink.api.java.utils.ParameterTool;
+
+import java.util.Base64;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * MainApp
@@ -35,11 +39,22 @@
  */
 public class MainApp {
 
-    public static void main(String[] args) throws IOException {
-        Map<String, String> params = FlinkBaseUtil.getParamsFromArgs(args);
-        String id = params.get(FlinkParamConstant.ID);
-        Asserts.checkNullString(id, "请配置入参 id ");
-        DBConfig dbConfig = DBConfig.build(params);
-        Submitter.submit(Integer.valueOf(id), dbConfig, params.get(FlinkParamConstant.DINKY_ADDR));
+    private static final Logger log = LoggerFactory.getLogger(Submitter.class);
+
+    public static void main(String[] args) throws Exception {
+        log.info("=========================Start run dinky app job===============================");
+        ParameterTool parameters = ParameterTool.fromArgs(args);
+        boolean isEncrypt = parameters.getBoolean(AppParamConstant.isEncrypt, true);
+        String config = parameters.get(AppParamConstant.config);
+        config = isEncrypt ? new String(Base64.getDecoder().decode(config)) : config;
+        AppParamConfig appConfig = JsonUtils.toJavaBean(config, AppParamConfig.class);
+        try {
+            log.info("dinky app is Ready to run, config is {}", appConfig);
+            DBUtil.init(appConfig);
+            Submitter.submit(appConfig);
+        } catch (Exception e) {
+            log.error("exectue app failed with config: {}", appConfig);
+            throw e;
+        }
     }
 }
diff --git a/dinky-app/dinky-app-base/pom.xml b/dinky-app/dinky-app-base/pom.xml
@@ -44,6 +44,10 @@
             <artifactId>lombok</artifactId>
             <scope>provided</scope>
         </dependency>
+        <dependency>
+            <groupId>cn.hutool</groupId>
+            <artifactId>hutool-all</artifactId>
+        </dependency>
     </dependencies>
 
     <profiles>

diff --git a/dinky-app/dinky-app-base/src/main/java/org/dinky/app/constant/AppParamConstant.java b/dinky-app/dinky-app-base/src/main/java/org/dinky/app/constant/AppParamConstant.java
@@ -0,0 +1,25 @@
+/*
+ *
+ *  Licensed to the Apache Software Foundation (ASF) under one or more
+ *  contributor license agreements.  See the NOTICE file distributed with
+ *  this work for additional information regarding copyright ownership.
+ *  The ASF licenses this file to You under the Apache License, Version 2.0
+ *  (the "License"); you may not use this file except in compliance with
+ *  the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ *
+ */
+
+package org.dinky.app.constant;
+
+public class AppParamConstant {
+    public static final String config = "config";
+    public static final String isEncrypt = "encrypt";
+}