[Feature-2989][dolphinscheduler] Dolphinscheduler supports common sql…

… and variables
aiwenmo · Jan 12, 2024 · 7d5327f · 7d5327f
1 parent 2b71dfe
commit 7d5327f
Show file tree

Hide file tree

Showing 8 changed files with 37 additions and 22 deletions.
diff --git a/dinky-admin/src/main/java/org/dinky/controller/APIController.java b/dinky-admin/src/main/java/org/dinky/controller/APIController.java
@@ -19,7 +19,9 @@
 
 package org.dinky.controller;
 
+import org.dinky.DinkyVersion;
 import org.dinky.data.annotations.Log;
+import org.dinky.data.dto.APISavePointTaskDTO;
 import org.dinky.data.dto.TaskDTO;
 import org.dinky.data.dto.TaskSubmitDto;
 import org.dinky.data.enums.BusinessType;
@@ -65,18 +67,10 @@ public class APIController {
     private final TaskService taskService;
     private final JobInstanceService jobInstanceService;
 
-    // Interface compatible with DolphinScheduler
-    @GetMapping("/submitTask")
-    @ApiOperation("Submit Task")
-    public Result<JobResult> submitTask(@RequestParam Integer id) throws Exception {
-        taskService.initTenantByTaskId(id);
-        JobResult jobResult =
-                taskService.submitTask(TaskSubmitDto.builder().id(id).build());
-        if (jobResult.isSuccess()) {
-            return Result.succeed(jobResult, Status.EXECUTE_SUCCESS);
-        } else {
-            return Result.failed(jobResult, jobResult.getError());
-        }
+    @GetMapping("/version")
+    @ApiOperation(value = "Query Service Version", notes = "Query Dinky Service Version Number")
+    public Result<String> getVersionInfo() {
+        return Result.succeed(DinkyVersion.getVersion(), "Get success");
     }
 
     @PostMapping("/submitTask")
@@ -92,6 +86,15 @@ public Result<JobResult> submitTask(@RequestBody TaskSubmitDto submitDto) throws
         }
     }
 
+    @PostMapping("/savepointTask")
+    public Result savepointTask(@RequestBody APISavePointTaskDTO apiSavePointTaskDTO) {
+        return Result.succeed(
+                taskService.savepointTaskJob(
+                        taskService.getTaskInfoById(apiSavePointTaskDTO.getTaskId()),
+                        SavePointType.get(apiSavePointTaskDTO.getType())),
+                Status.EXECUTE_SUCCESS);
+    }
+
     @GetMapping("/cancel")
     //    @Log(title = "Cancel Flink Job", businessType = BusinessType.TRIGGER)
     @ApiOperation("Cancel Flink Job")

diff --git a/dinky-admin/src/main/java/org/dinky/data/dto/TaskSubmitDto.java b/dinky-admin/src/main/java/org/dinky/data/dto/TaskSubmitDto.java
@@ -28,10 +28,25 @@
 @Data
 @Builder
 public class TaskSubmitDto {
+    public TaskSubmitDto() {}
+
+    public TaskSubmitDto(Integer id, Boolean isOnline, String savePointPath, Map<String, String> variables) {
+        this.id = id;
+        this.isOnline = isOnline;
+        this.savePointPath = savePointPath;
+        this.variables = variables;
+    }
 
     @ApiModelProperty(value = "ID", dataType = "Integer", example = "6", notes = "The identifier of the execution")
     private Integer id;
 
+    @ApiModelProperty(
+            value = "Is online",
+            dataType = "Boolean",
+            example = "true",
+            notes = "Online dinky task, and only one job is allowed to execute")
+    private Boolean isOnline;
+
     @ApiModelProperty(
             value = "Save Point Path",
             dataType = "String",

diff --git a/dinky-admin/src/main/java/org/dinky/init/SystemInit.java b/dinky-admin/src/main/java/org/dinky/init/SystemInit.java
@@ -202,9 +202,7 @@ private void aboutDolphinSchedulerInitOperation(Object v) {
                     project = projectClient.createDinkyProject();
                 }
             } catch (Exception e) {
-                log.error("Error in DolphinScheduler: ", e);
-                log.error(
-                        "get or create DolphinScheduler project failed, please check the config of DolphinScheduler!");
+                log.warn("Get or create DolphinScheduler project failed, please check the config of DolphinScheduler!");
             }
         }
     }

diff --git a/...adata/dinky-metadata-base/src/main/java/org/dinky/metadata/driver/AbstractJdbcDriver.java b/...adata/dinky-metadata-base/src/main/java/org/dinky/metadata/driver/AbstractJdbcDriver.java
@@ -600,8 +600,9 @@ public JdbcSelectResult query(String sql, Integer limit) {
         } catch (Exception e) {
             result.error(LogUtil.getError(e));
             log.error("Query failed", e);
+        } finally {
+            close(preparedStatement, results);
         }
-        close(preparedStatement, results);
         result.setRowData(datas);
         return result;
     }

diff --git a/dinky-scheduler/src/main/java/org/dinky/scheduler/model/DinkyTaskParams.java b/dinky-scheduler/src/main/java/org/dinky/scheduler/model/DinkyTaskParams.java
@@ -19,6 +19,7 @@
 
 package org.dinky.scheduler.model;
 
+import java.util.ArrayList;
 import java.util.List;
 
 import io.swagger.annotations.ApiModelProperty;
@@ -28,7 +29,7 @@
 public class DinkyTaskParams {
 
     @ApiModelProperty(value = "自定义参数")
-    private List<Property> localParams;
+    private List<Property> localParams = new ArrayList<>();
 
     @ApiModelProperty(value = "dinky地址")
     private String address;

diff --git a/dinky-web/src/pages/DataStudio/HeaderContainer/function.tsx b/dinky-web/src/pages/DataStudio/HeaderContainer/function.tsx
@@ -46,7 +46,6 @@ export const isOnline = (data: TaskDataType | undefined) => {
 export const isCanPushDolphin = (data: TaskDataType | undefined) => {
   return data
     ? JOB_LIFE_CYCLE.PUBLISH === data.step &&
-        !isSql(data?.dialect) &&
         data?.dialect?.toLowerCase() !== DIALECT.FLINKSQLENV &&
         data?.dialect?.toLowerCase() !== DIALECT.SCALA &&
         data?.dialect?.toLowerCase() !== DIALECT.JAVA &&

diff --git a/dinky-web/src/pages/DataStudio/HeaderContainer/index.tsx b/dinky-web/src/pages/DataStudio/HeaderContainer/index.tsx
@@ -367,9 +367,7 @@ const HeaderContainer = (props: connect) => {
       icon: isOnline(currentData) ? <MergeCellsOutlined /> : <FundOutlined />,
       title: isOnline(currentData) ? l('button.offline') : l('button.publish'),
       isShow:
-        (currentTab?.type == TabsPageType.project &&
-          currentTab?.subType?.toLowerCase() === DIALECT.FLINK_SQL) ||
-        currentTab?.subType?.toLowerCase() === DIALECT.FLINKJAR,
+        (currentTab?.type == TabsPageType.project),
       click: () => handleChangeJobLife()
     },
     {

diff --git a/pom.xml b/pom.xml
@@ -607,7 +607,7 @@
             </dependency>
             <dependency>
                 <groupId>org.dinky</groupId>
-                <artifactId>dinky-cdc-doris</artifactId>
+                <artifactId>dinky-cdc-plus</artifactId>
                 <version>${project.version}</version>
             </dependency>
             <dependency>