[Improve][feature] Add JobInstance Delete

pom.xml

@@ -71,11 +71,12 @@
         <guava.version>19.0</guava.version>
         <checker.qual.version>3.10.0</checker.qual.version>
         <awaitility.version>4.2.0</awaitility.version>
-        <seatunnel-framework.version>2.3.3</seatunnel-framework.version>
+        <seatunnel-framework.version>2.3.4</seatunnel-framework.version>
         <oracle-jdbc.version>21.5.0.0</oracle-jdbc.version>
         <postgresql.version>42.4.3</postgresql.version>
         <sqlserver.version>9.2.1.jre8</sqlserver.version>
         <mysql-connector.version>8.0.28</mysql-connector.version>
+        <clickhouse.version>0.4.6</clickhouse.version>
         <hadoop-aws.version>3.1.4</hadoop-aws.version>
         <aws-java-sdk-bundle.version>1.11.271</aws-java-sdk-bundle.version>
         <auto-service.version>1.0.1</auto-service.version>

seatunnel-datasource/seatunnel-datasource-plugins/datasource-jdbc-db2/src/main/java/org/apache/seatunnel/datasource/plugin/db2/jdbc/Db2JdbcDataSourceChannel.java

@@ -1,3 +1,20 @@
+/*
+ * 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.apache.seatunnel.datasource.plugin.db2.jdbc;
 
 import org.apache.seatunnel.api.configuration.util.OptionRule;

seatunnel-server/seatunnel-app/pom.xml

@@ -479,14 +479,14 @@
             <groupId>mysql</groupId>
             <artifactId>mysql-connector-java</artifactId>
             <version>${mysql.version}</version>
-            <scope>test</scope>
+            <!--            <scope>test</scope>-->
         </dependency>
 
         <dependency>
             <groupId>com.oracle.database.jdbc</groupId>
             <artifactId>ojdbc8</artifactId>
             <version>${oracle-jdbc.version}</version>
-            <scope>test</scope>
+            <!--            <scope>test</scope>-->
         </dependency>
 
         <dependency>
@@ -503,6 +503,12 @@
             <scope>test</scope>
         </dependency>
 
+        <dependency>
+            <groupId>com.clickhouse</groupId>
+            <artifactId>clickhouse-jdbc</artifactId>
+            <version>${clickhouse.version}</version>
+        </dependency>
+
     </dependencies>
 
     <build>
@@ -532,5 +538,4 @@
             </plugin>
         </plugins>
     </build>
-
 </project>

seatunnel-server/seatunnel-app/src/main/java/org/apache/seatunnel/app/bean/engine/EngineDataType.java

@@ -16,7 +16,6 @@
  */
 package org.apache.seatunnel.app.bean.engine;
 
-import org.apache.seatunnel.api.table.catalog.DataTypeConvertException;
 import org.apache.seatunnel.api.table.catalog.DataTypeConvertor;
 import org.apache.seatunnel.api.table.type.ArrayType;
 import org.apache.seatunnel.api.table.type.BasicType;
@@ -50,6 +49,12 @@ public enum DataType {
         T_BYTE("tinyint", BasicType.BYTE_TYPE),
         T_SHORT("smallint", BasicType.SHORT_TYPE),
         T_INT("int", BasicType.INT_TYPE),
+
+        T_INT8("int8", BasicType.INT_TYPE),
+        T_UNIT32("uint32", BasicType.LONG_TYPE),
+
+        T_Float32("Float32", BasicType.FLOAT_TYPE),
+
         T_LONG("bigint", BasicType.LONG_TYPE),
         T_FLOAT("float", BasicType.FLOAT_TYPE),
         T_DOUBLE("double", BasicType.DOUBLE_TYPE),
@@ -86,21 +91,19 @@ public static class SeaTunnelDataTypeConvertor
             implements DataTypeConvertor<SeaTunnelDataType<?>> {
 
         @Override
-        public SeaTunnelDataType<?> toSeaTunnelType(String engineDataType) {
-            return DATA_TYPE_MAP.get(engineDataType.toLowerCase(Locale.ROOT)).getRawType();
+        public SeaTunnelDataType<?> toSeaTunnelType(String s, String s1) {
+            return DATA_TYPE_MAP.get(s.toLowerCase(Locale.ROOT)).getRawType();
         }
 
         @Override
         public SeaTunnelDataType<?> toSeaTunnelType(
-                SeaTunnelDataType<?> seaTunnelDataType, Map<String, Object> map)
-                throws DataTypeConvertException {
+                String s, SeaTunnelDataType<?> seaTunnelDataType, Map<String, Object> map) {
             return seaTunnelDataType;
         }
 
         @Override
         public SeaTunnelDataType<?> toConnectorType(
-                SeaTunnelDataType<?> seaTunnelDataType, Map<String, Object> map)
-                throws DataTypeConvertException {
+                String s, SeaTunnelDataType<?> seaTunnelDataType, Map<String, Object> map) {
             return seaTunnelDataType;
         }
 

seatunnel-server/seatunnel-app/src/main/java/org/apache/seatunnel/app/controller/JobExecutorController.java

@@ -18,13 +18,11 @@
 package org.apache.seatunnel.app.controller;
 
 import org.apache.seatunnel.app.common.Result;
-import org.apache.seatunnel.app.domain.response.executor.JobExecutorRes;
 import org.apache.seatunnel.app.service.IJobExecutorService;
-import org.apache.seatunnel.app.service.IJobInstanceService;
-import org.apache.seatunnel.server.common.SeatunnelErrorEnum;
-import org.apache.seatunnel.server.common.SeatunnelException;
 
+import org.springframework.web.bind.annotation.DeleteMapping;
 import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.bind.annotation.RequestAttribute;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RequestParam;
@@ -44,34 +42,26 @@
 public class JobExecutorController {
 
     @Resource IJobExecutorService jobExecutorService;
-    @Resource private IJobInstanceService jobInstanceService;
 
-    @GetMapping("/execute")
-    @ApiOperation(value = "Execute synchronization tasks", httpMethod = "GET")
-    public Result<Long> jobExecutor(
+    @PostMapping("/create")
+    @ApiOperation(value = "Create synchronization tasks", httpMethod = "POST")
+    public Result<Long> jobCreate(
             @ApiParam(value = "userId", required = true) @RequestAttribute("userId") Integer userId,
             @ApiParam(value = "jobDefineId", required = true) @RequestParam("jobDefineId")
-                    Long jobDefineId) {
-        return jobExecutorService.jobExecute(userId, jobDefineId);
+            Long jobDefineId) {
+        return jobExecutorService.jobCreate(userId, jobDefineId);
     }
 
-    @GetMapping("/resource")
-    @ApiOperation(value = "get the resource for job executor", httpMethod = "GET")
-    public Result<JobExecutorRes> resource(
-            @ApiParam(value = "userId", required = true) @RequestParam Integer userId,
-            @ApiParam(value = "Job define id", required = true) @RequestParam Long jobDefineId)
-            throws IOException {
-        try {
-            JobExecutorRes executeResource =
-                    jobInstanceService.createExecuteResource(userId, jobDefineId);
-            return Result.success(executeResource);
-        } catch (Exception e) {
-            log.error("Get the resource for job executor error", e);
-            throw new SeatunnelException(SeatunnelErrorEnum.ILLEGAL_STATE, e.getMessage());
-        }
+    @PostMapping("/execute")
+    @ApiOperation(value = "Execute synchronization tasks", httpMethod = "Post")
+    public Result<Long> jobExecutor(
+            @ApiParam(value = "userId", required = true) @RequestAttribute("userId") Integer userId,
+            @ApiParam(value = "jobInstanceId", required = true) @RequestParam Long jobInstanceId) {
+        return jobExecutorService.jobExecute(userId, jobInstanceId);
     }
 
-    @GetMapping("/pause")
+    @PostMapping("/pause")
+    @ApiOperation(value = "Pause synchronization tasks", httpMethod = "POST")
     public Result<Void> jobPause(
             @ApiParam(value = "userId", required = true) @RequestAttribute("userId") Integer userId,
             @ApiParam(value = "jobInstanceId", required = true) @RequestParam Long jobInstanceId) {
@@ -84,4 +74,12 @@ public Result<Void> jobRestore(
             @ApiParam(value = "jobInstanceId", required = true) @RequestParam Long jobInstanceId) {
         return jobExecutorService.jobStore(userId, jobInstanceId);
     }
+
+    @DeleteMapping("/delete")
+    @ApiOperation(value = "Delete synchronization tasks", httpMethod = "DELETE")
+    public Result<Void> jobCancel(
+            @ApiParam(value = "userId", required = true) @RequestAttribute("userId") Integer userId,
+            @ApiParam(value = "jobInstanceId", required = true) @RequestParam Long jobInstanceId) {
+        return jobExecutorService.jobDelete(userId, jobInstanceId);
+    }
 }

seatunnel-server/seatunnel-app/src/main/java/org/apache/seatunnel/app/dal/dao/IJobInstanceDao.java

@@ -37,6 +37,8 @@ public interface IJobInstanceDao {
 
     void insert(@NonNull JobInstance jobInstance);
 
+    void delete(@NonNull JobInstance jobInstance);
+
     JobInstanceMapper getJobInstanceMapper();
 
     IPage<SeaTunnelJobInstanceDto> queryJobInstanceListPaging(

seatunnel-server/seatunnel-app/src/main/java/org/apache/seatunnel/app/dal/dao/impl/JobInstanceDaoImpl.java

@@ -60,6 +60,11 @@ public void insert(@NonNull JobInstance jobInstance) {
         jobInstanceMapper.insert(jobInstance);
     }
 
+    @Override
+    public void delete(@NonNull JobInstance jobInstance) {
+        jobInstanceMapper.deleteById(jobInstance);
+    }
+
     @Override
     public JobInstanceMapper getJobInstanceMapper() {
         return jobInstanceMapper;

seatunnel-server/seatunnel-app/src/main/java/org/apache/seatunnel/app/permission/constants/SeatunnelFuncPermissionKeyConstant.java

@@ -53,9 +53,13 @@ public class SeatunnelFuncPermissionKeyConstant {
     public static final String JOB_TABLE_SCHEMA = "project:seatunnel-task:table-schema";
     public static final String JOB_TABLE_COLUMN_PROJECTION =
             "project:seatunnel-task:column-projection";
-    public static final String JOB_EXECUTOR_RESOURCE = "project:seatunnel-task:job-exec-resource";
+    public static final String JOB_EXECUTOR_CREATE = "project:seatunnel-task:job-exec-create";
     public static final String JOB_EXECUTOR_INSTANCE = "project:seatunnel-task:job-exec-instance";
+    public static final String JOB_EXECUTOR_STATUS = "project:seatunnel-task:job-exec-status";
+    public static final String JOB_EXECUTOR_START = "project:seatunnel-task:job-exec-start";
+    public static final String JOB_EXECUTOR_PAUSE = "project:seatunnel-task:job-exec-pause";
     public static final String JOB_EXECUTOR_COMPLETE = "project:seatunnel-task:job-exec-complete";
+    public static final String JOB_EXECUTOR_DELETE = "project:seatunnel-task:job-exec-delete";
 
     /** sync task instance */
     public static final String JOB_METRICS_SUMMARY = "project:seatunnel-task-instance:summary";

seatunnel-server/seatunnel-app/src/main/java/org/apache/seatunnel/app/service/IJobExecutorService.java

@@ -21,9 +21,13 @@
 
 public interface IJobExecutorService {
 
-    Result<Long> jobExecute(Integer userId, Long jobDefineId);
+    Result<Long> jobCreate(Integer userId, Long jobDefineId);
+
+    Result<Long> jobExecute(Integer userId, Long jobInstanceId);
 
     Result<Void> jobPause(Integer userId, Long jobInstanceId);
 
     Result<Void> jobStore(Integer userId, Long jobInstanceId);
+
+    Result<Void> jobDelete(Integer userId, Long jobInstanceId);
 }

seatunnel-server/seatunnel-app/src/main/java/org/apache/seatunnel/app/service/IJobInstanceService.java

@@ -17,21 +17,32 @@
 
 package org.apache.seatunnel.app.service;
 
+import org.apache.seatunnel.app.dal.entity.JobInstance;
 import org.apache.seatunnel.app.dal.entity.JobLine;
 import org.apache.seatunnel.app.dal.entity.JobTask;
-import org.apache.seatunnel.app.domain.response.executor.JobExecutorRes;
 
 import lombok.NonNull;
+import org.apache.seatunnel.engine.client.job.ClientJobProxy;
+import org.apache.seatunnel.engine.core.job.JobStatus;
 
 import java.util.List;
 
 public interface IJobInstanceService {
-    JobExecutorRes createExecuteResource(@NonNull Integer userId, @NonNull Long jobDefineId);
-
     String generateJobConfig(Long jobId, List<JobTask> tasks, List<JobLine> lines, String envStr);
 
-    JobExecutorRes getExecuteResource(@NonNull Long jobEngineId);
+    Long create(@NonNull Integer userId, @NonNull Long jobDefineId);
+
+    void start(@NonNull Integer userId, @NonNull Long jobInstanceId);
+
+    JobStatus status(@NonNull Integer userId, @NonNull Long jobInstanceId);
+
+    void pause(@NonNull Integer userId, @NonNull Long jobInstanceId);
+
+    void restore(@NonNull Integer userId, @NonNull Long jobInstanceId);
+
+    void delete(@NonNull Integer userId, @NonNull Long jobInstanceId);
+
+    void complete(@NonNull Integer userId, @NonNull Long jobInstanceId, @NonNull String jobEngineId);
 
-    void complete(
-            @NonNull Integer userId, @NonNull Long jobInstanceId, @NonNull String jobEngineId);
+    void delete(@NonNull Integer userId, @NonNull Long jobInstanceId, @NonNull String jobEngineId);
 }

seatunnel-server/seatunnel-app/src/main/java/org/apache/seatunnel/app/service/IJobMetricsService.java

@@ -47,6 +47,9 @@ JobDAG getJobDAG(@NonNull Integer userId, @NonNull Long jobInstanceId)
     void syncJobDataToDb(
             @NonNull JobInstance jobInstance, @NonNull Integer userId, @NonNull String jobEngineId);
 
+    void deleteJobDataToDb(
+            @NonNull JobInstance jobInstance, @NonNull Integer userId, @NonNull String jobEngineId);
+
     JobSummaryMetricsRes getJobSummaryMetrics(
             @NonNull Integer userId, @NonNull Long jobInstanceId, @NonNull String jobEngineId);