diff --git a/dinky-admin/src/main/java/org/dinky/service/TaskService.java b/dinky-admin/src/main/java/org/dinky/service/TaskService.java index 5fc47a370f..7a68610be5 100644 --- a/dinky-admin/src/main/java/org/dinky/service/TaskService.java +++ b/dinky-admin/src/main/java/org/dinky/service/TaskService.java @@ -35,6 +35,7 @@ import org.dinky.explainer.lineage.LineageResult; import org.dinky.gateway.enums.SavePointType; import org.dinky.gateway.result.SavePointResult; +import org.dinky.job.JobConfig; import org.dinky.job.JobResult; import org.dinky.mybatis.service.ISuperService; @@ -281,4 +282,11 @@ public interface TaskService extends ISuperService { * @return A {@link LineageResult} object representing the found task lineage. */ LineageResult getTaskLineage(Integer id); + + /** + * Build the job submit config with the given task + * @param task + * @return + */ + JobConfig buildJobSubmitConfig(TaskDTO task); } 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 index fbd4b48212..1c3a95e0ae 100644 --- a/dinky-admin/src/main/java/org/dinky/service/impl/TaskServiceImpl.java +++ b/dinky-admin/src/main/java/org/dinky/service/impl/TaskServiceImpl.java @@ -203,7 +203,7 @@ public JobResult executeJob(TaskDTO task, Boolean stream) throws Exception { return jobResult; } - // Submit and export task + @Override @ProcessStep(type = ProcessStepType.SUBMIT_BUILD_CONFIG) public JobConfig buildJobSubmitConfig(TaskDTO task) { if (Asserts.isNull(task.getType())) { diff --git a/dinky-admin/src/main/java/org/dinky/service/task/FlinkSqlEnvTask.java b/dinky-admin/src/main/java/org/dinky/service/task/FlinkSqlEnvTask.java index 8168e61cd0..d0aa860d47 100644 --- a/dinky-admin/src/main/java/org/dinky/service/task/FlinkSqlEnvTask.java +++ b/dinky-admin/src/main/java/org/dinky/service/task/FlinkSqlEnvTask.java @@ -29,6 +29,7 @@ import org.dinky.gateway.enums.GatewayType; import org.dinky.job.JobManager; import org.dinky.job.JobResult; +import org.dinky.service.TaskService; import org.dinky.service.impl.TaskServiceImpl; import java.util.List; @@ -69,7 +70,7 @@ public boolean stop() { } protected JobManager getJobManager() { - TaskServiceImpl taskService = SpringUtil.getBean(TaskServiceImpl.class); + TaskService taskService = SpringUtil.getBean(TaskServiceImpl.class); return JobManager.build(taskService.buildJobSubmitConfig(task)); } } diff --git a/dinky-admin/src/main/java/org/dinky/service/task/FlinkSqlTask.java b/dinky-admin/src/main/java/org/dinky/service/task/FlinkSqlTask.java index c11550228b..c4f4d1ee1d 100644 --- a/dinky-admin/src/main/java/org/dinky/service/task/FlinkSqlTask.java +++ b/dinky-admin/src/main/java/org/dinky/service/task/FlinkSqlTask.java @@ -27,6 +27,7 @@ import org.dinky.gateway.enums.GatewayType; import org.dinky.job.JobManager; import org.dinky.job.JobResult; +import org.dinky.service.TaskService; import org.dinky.service.impl.TaskServiceImpl; import org.dinky.utils.JsonUtils; @@ -68,7 +69,7 @@ public JobResult execute() throws Exception { } protected JobManager getJobManager() { - TaskServiceImpl taskService = SpringUtil.getBean(TaskServiceImpl.class); + TaskService taskService = SpringUtil.getBean(TaskServiceImpl.class); return JobManager.build(taskService.buildJobSubmitConfig(task)); }