This is an automated email from the ASF dual-hosted git repository.
nicholasjiang pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-webui.git
The following commit(s) were added to refs/heads/main by this push:
new 6981175 [Feature] Introduce job submission interface (#222)
6981175 is described below
commit 698117546d784de7c79709c8afd4676f10cac54d
Author: s7monk <[email protected]>
AuthorDate: Thu May 16 15:37:16 2024 +0800
[Feature] Introduce job submission interface (#222)
---
.../web/server/controller/JobController.java | 120 ++++++
.../paimon/web/server/data/dto/JobSubmitDTO.java | 40 ++
.../paimon/web/server/data/dto/ResultFetchDTO.java | 36 ++
.../paimon/web/server/data/dto/StopJobDTO.java | 34 ++
.../paimon/web/server/data/model/JobInfo.java | 62 +++
.../web/server/data/result/enums/Status.java | 8 +-
.../paimon/web/server/data/vo/JobStatisticsVO.java | 42 ++
.../paimon/web/server/data/vo/JobStatusVO.java | 36 ++
.../apache/paimon/web/server/data/vo/JobVO.java | 64 +++
.../paimon/web/server/data/vo/ResultDataVO.java | 43 ++
.../apache/paimon/web/server/mapper/JobMapper.java | 28 ++
.../web/server/service/JobExecutorService.java | 40 ++
.../paimon/web/server/service/JobService.java | 96 +++++
.../web/server/service/impl/JobServiceImpl.java | 478 +++++++++++++++++++++
.../paimon/web/server/util/LocalDateTimeUtil.java | 36 ++
.../src/main/resources/i18n/messages.properties | 4 +
.../main/resources/i18n/messages_en_US.properties | 4 +
.../main/resources/i18n/messages_zh_CN.properties | 4 +
.../web/server/constant/StatementsConstant.java | 76 ++++
.../web/server/controller/JobControllerTest.java | 404 +++++++++++++++++
scripts/sql/paimon-mysql.sql | 18 +
21 files changed, 1672 insertions(+), 1 deletion(-)
diff --git
a/paimon-web-server/src/main/java/org/apache/paimon/web/server/controller/JobController.java
b/paimon-web-server/src/main/java/org/apache/paimon/web/server/controller/JobController.java
new file mode 100644
index 0000000..a7be95d
--- /dev/null
+++
b/paimon-web-server/src/main/java/org/apache/paimon/web/server/controller/JobController.java
@@ -0,0 +1,120 @@
+/*
+ * 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.paimon.web.server.controller;
+
+import org.apache.paimon.web.server.data.dto.JobSubmitDTO;
+import org.apache.paimon.web.server.data.dto.ResultFetchDTO;
+import org.apache.paimon.web.server.data.dto.StopJobDTO;
+import org.apache.paimon.web.server.data.model.JobInfo;
+import org.apache.paimon.web.server.data.result.R;
+import org.apache.paimon.web.server.data.result.enums.Status;
+import org.apache.paimon.web.server.data.vo.JobStatisticsVO;
+import org.apache.paimon.web.server.data.vo.JobStatusVO;
+import org.apache.paimon.web.server.data.vo.JobVO;
+import org.apache.paimon.web.server.data.vo.ResultDataVO;
+import org.apache.paimon.web.server.service.JobService;
+
+import cn.dev33.satoken.annotation.SaCheckPermission;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.util.List;
+
+/** Job submit api controller. */
+@Slf4j
+@RestController
+@RequestMapping("/api/job")
+public class JobController {
+
+ @Autowired private JobService jobService;
+
+ @SaCheckPermission("playground:job:submit")
+ @PostMapping("/submit")
+ public R<JobVO> submit(@RequestBody JobSubmitDTO jobSubmitDTO) {
+ try {
+ return R.succeed(jobService.submitJob(jobSubmitDTO));
+ } catch (Exception e) {
+ log.error("Exception with submitting a job.", e);
+ return R.failed(Status.JOB_SUBMIT_ERROR);
+ }
+ }
+
+ @SaCheckPermission("playground:job:fetch")
+ @PostMapping("/fetch")
+ public R<ResultDataVO> fetchResult(@RequestBody ResultFetchDTO
resultFetchDTO) {
+ try {
+ return R.succeed(jobService.fetchResult(resultFetchDTO));
+ } catch (Exception e) {
+ log.error("Exception with fetching result data.", e);
+ return R.failed(Status.RESULT_FETCH_ERROR);
+ }
+ }
+
+ @SaCheckPermission("playground:job:list")
+ @GetMapping("/list")
+ public R<List<JobVO>> list() {
+ return R.succeed(jobService.listJobs());
+ }
+
+ @SaCheckPermission("playground:job:list")
+ @GetMapping("/list/page")
+ public R<List<JobVO>> listJobsByPage(int current, int size) {
+ return R.succeed(jobService.listJobsByPage(current, size));
+ }
+
+ @SaCheckPermission("playground:job:query")
+ @GetMapping("/status/get/{jobId}")
+ public R<JobStatusVO> getJobStatus(@PathVariable("jobId") String jobId) {
+ JobInfo job = jobService.getJobById(jobId);
+ JobStatusVO jobStatusVO =
+
JobStatusVO.builder().jobId(job.getJobId()).status(job.getStatus()).build();
+ return R.succeed(jobStatusVO);
+ }
+
+ @SaCheckPermission("playground:job:query")
+ @GetMapping("/statistics/get")
+ public R<JobStatisticsVO> getJobStatistics() {
+ return R.succeed(jobService.getJobStatistics());
+ }
+
+ @SaCheckPermission("playground:job:stop")
+ @PostMapping("/stop")
+ public R<Void> stop(@RequestBody StopJobDTO stopJobDTO) {
+ try {
+ jobService.stop(stopJobDTO);
+ return R.succeed();
+ } catch (Exception e) {
+ log.error("Exception with stopping a job.", e);
+ return R.failed(Status.JOB_STOP_ERROR);
+ }
+ }
+
+ @SaCheckPermission("playground:job:refresh")
+ @PostMapping("/refresh")
+ public R<Void> refresh() {
+ jobService.refreshJobStatus("Flink");
+ return R.succeed();
+ }
+}
diff --git
a/paimon-web-server/src/main/java/org/apache/paimon/web/server/data/dto/JobSubmitDTO.java
b/paimon-web-server/src/main/java/org/apache/paimon/web/server/data/dto/JobSubmitDTO.java
new file mode 100644
index 0000000..a8cce8b
--- /dev/null
+++
b/paimon-web-server/src/main/java/org/apache/paimon/web/server/data/dto/JobSubmitDTO.java
@@ -0,0 +1,40 @@
+/*
+ * 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.paimon.web.server.data.dto;
+
+import lombok.Data;
+
+import java.util.Map;
+
+/** DTO of submitting job. */
+@Data
+public class JobSubmitDTO {
+
+ private String jobName;
+
+ private String taskType;
+
+ private boolean isStreaming;
+
+ private String clusterId;
+
+ private Map<String, String> config;
+
+ private String statements;
+}
diff --git
a/paimon-web-server/src/main/java/org/apache/paimon/web/server/data/dto/ResultFetchDTO.java
b/paimon-web-server/src/main/java/org/apache/paimon/web/server/data/dto/ResultFetchDTO.java
new file mode 100644
index 0000000..1d34e6f
--- /dev/null
+++
b/paimon-web-server/src/main/java/org/apache/paimon/web/server/data/dto/ResultFetchDTO.java
@@ -0,0 +1,36 @@
+/*
+ * 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.paimon.web.server.data.dto;
+
+import lombok.Data;
+
+/** DTO of fetching result. */
+@Data
+public class ResultFetchDTO {
+
+ private String submitId;
+
+ private String clusterId;
+
+ private String sessionId;
+
+ private String taskType;
+
+ private Long token;
+}
diff --git
a/paimon-web-server/src/main/java/org/apache/paimon/web/server/data/dto/StopJobDTO.java
b/paimon-web-server/src/main/java/org/apache/paimon/web/server/data/dto/StopJobDTO.java
new file mode 100644
index 0000000..91430df
--- /dev/null
+++
b/paimon-web-server/src/main/java/org/apache/paimon/web/server/data/dto/StopJobDTO.java
@@ -0,0 +1,34 @@
+/*
+ * 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.paimon.web.server.data.dto;
+
+import lombok.Data;
+
+/** DTO of stopping job. */
+@Data
+public class StopJobDTO {
+
+ private String clusterId;
+
+ private String jobId;
+
+ private String taskType;
+
+ private boolean withSavepoint;
+}
diff --git
a/paimon-web-server/src/main/java/org/apache/paimon/web/server/data/model/JobInfo.java
b/paimon-web-server/src/main/java/org/apache/paimon/web/server/data/model/JobInfo.java
new file mode 100644
index 0000000..ea05d14
--- /dev/null
+++
b/paimon-web-server/src/main/java/org/apache/paimon/web/server/data/model/JobInfo.java
@@ -0,0 +1,62 @@
+/*
+ * 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.paimon.web.server.data.model;
+
+import com.baomidou.mybatisplus.annotation.TableName;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.NoArgsConstructor;
+
+import java.time.LocalDateTime;
+
+/** Job table model. */
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+@EqualsAndHashCode(callSuper = true)
+@TableName("job")
+public class JobInfo extends BaseModel {
+
+ private String jobId;
+
+ private String jobName;
+
+ private String type;
+
+ private String executeMode;
+
+ private String clusterId;
+
+ private Integer uid;
+
+ private String config;
+
+ private String statements;
+
+ private String status;
+
+ private LocalDateTime startTime;
+
+ private LocalDateTime endTime;
+
+ private static final long serialVersionUID = 1L;
+}
diff --git
a/paimon-web-server/src/main/java/org/apache/paimon/web/server/data/result/enums/Status.java
b/paimon-web-server/src/main/java/org/apache/paimon/web/server/data/result/enums/Status.java
index 29baaa1..4ebb130 100644
---
a/paimon-web-server/src/main/java/org/apache/paimon/web/server/data/result/enums/Status.java
+++
b/paimon-web-server/src/main/java/org/apache/paimon/web/server/data/result/enums/Status.java
@@ -83,7 +83,13 @@ public enum Status {
/** ------------cluster-----------------. */
CLUSTER_NOT_EXIST(10701, "cluster.not.exist"),
- CLUSTER_NAME_ALREADY_EXISTS(10702, "cluster.name.exist");
+ CLUSTER_NAME_ALREADY_EXISTS(10702, "cluster.name.exist"),
+
+ /** ------------job-----------------. */
+ JOB_SUBMIT_ERROR(10701, "job.submit.error"),
+ RESULT_FETCH_ERROR(10702, "result.fetch.error"),
+ JOB_STOP_ERROR(10703, "job.stop.error"),
+ JOB_UPDATE_STATUS_ERROR(10704, "job.update.status.error");
private final int code;
private final String msg;
diff --git
a/paimon-web-server/src/main/java/org/apache/paimon/web/server/data/vo/JobStatisticsVO.java
b/paimon-web-server/src/main/java/org/apache/paimon/web/server/data/vo/JobStatisticsVO.java
new file mode 100644
index 0000000..802acf5
--- /dev/null
+++
b/paimon-web-server/src/main/java/org/apache/paimon/web/server/data/vo/JobStatisticsVO.java
@@ -0,0 +1,42 @@
+/*
+ * 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.paimon.web.server.data.vo;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/** VO of JobStatistics. */
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class JobStatisticsVO {
+
+ private Long totalNum;
+
+ private Long runningNum;
+
+ private Long finishedNum;
+
+ private Long canceledNum;
+
+ private Long failedNum;
+}
diff --git
a/paimon-web-server/src/main/java/org/apache/paimon/web/server/data/vo/JobStatusVO.java
b/paimon-web-server/src/main/java/org/apache/paimon/web/server/data/vo/JobStatusVO.java
new file mode 100644
index 0000000..e36cf20
--- /dev/null
+++
b/paimon-web-server/src/main/java/org/apache/paimon/web/server/data/vo/JobStatusVO.java
@@ -0,0 +1,36 @@
+/*
+ * 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.paimon.web.server.data.vo;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/** VO of JobStatus. */
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class JobStatusVO {
+
+ private String jobId;
+
+ private String status;
+}
diff --git
a/paimon-web-server/src/main/java/org/apache/paimon/web/server/data/vo/JobVO.java
b/paimon-web-server/src/main/java/org/apache/paimon/web/server/data/vo/JobVO.java
new file mode 100644
index 0000000..5e22b96
--- /dev/null
+++
b/paimon-web-server/src/main/java/org/apache/paimon/web/server/data/vo/JobVO.java
@@ -0,0 +1,64 @@
+/*
+ * 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.paimon.web.server.data.vo;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.time.LocalDateTime;
+import java.util.List;
+import java.util.Map;
+
+/** VO of job. */
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class JobVO {
+
+ private String submitId;
+
+ private String jobId;
+
+ private String jobName;
+
+ private String type;
+
+ private String executeMode;
+
+ private String clusterId;
+
+ private String sessionId;
+
+ private Integer uid;
+
+ private String status;
+
+ private Boolean shouldFetchResult;
+
+ private List<Map<String, Object>> resultData;
+
+ private Long token;
+
+ private LocalDateTime startTime;
+
+ private LocalDateTime endTime;
+}
diff --git
a/paimon-web-server/src/main/java/org/apache/paimon/web/server/data/vo/ResultDataVO.java
b/paimon-web-server/src/main/java/org/apache/paimon/web/server/data/vo/ResultDataVO.java
new file mode 100644
index 0000000..097d28b
--- /dev/null
+++
b/paimon-web-server/src/main/java/org/apache/paimon/web/server/data/vo/ResultDataVO.java
@@ -0,0 +1,43 @@
+/*
+ * 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.paimon.web.server.data.vo;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.util.List;
+import java.util.Map;
+
+/** VO of result data. */
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class ResultDataVO {
+
+ private List<Map<String, Object>> resultData;
+
+ private Integer columns;
+
+ private Integer rows;
+
+ private Long token;
+}
diff --git
a/paimon-web-server/src/main/java/org/apache/paimon/web/server/mapper/JobMapper.java
b/paimon-web-server/src/main/java/org/apache/paimon/web/server/mapper/JobMapper.java
new file mode 100644
index 0000000..2440fab
--- /dev/null
+++
b/paimon-web-server/src/main/java/org/apache/paimon/web/server/mapper/JobMapper.java
@@ -0,0 +1,28 @@
+/*
+ * 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.paimon.web.server.mapper;
+
+import org.apache.paimon.web.server.data.model.JobInfo;
+
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import org.apache.ibatis.annotations.Mapper;
+
+/** Job mapper. */
+@Mapper
+public interface JobMapper extends BaseMapper<JobInfo> {}
diff --git
a/paimon-web-server/src/main/java/org/apache/paimon/web/server/service/JobExecutorService.java
b/paimon-web-server/src/main/java/org/apache/paimon/web/server/service/JobExecutorService.java
new file mode 100644
index 0000000..db60d1f
--- /dev/null
+++
b/paimon-web-server/src/main/java/org/apache/paimon/web/server/service/JobExecutorService.java
@@ -0,0 +1,40 @@
+/*
+ * 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.paimon.web.server.service;
+
+import org.apache.paimon.web.engine.flink.common.executor.Executor;
+
+import org.springframework.stereotype.Service;
+
+import java.util.concurrent.ConcurrentHashMap;
+
+/** Job executor service. */
+@Service
+public class JobExecutorService {
+
+ private final ConcurrentHashMap<String, Executor> executors = new
ConcurrentHashMap<>();
+
+ public Executor getExecutor(String sessionId) {
+ return executors.get(sessionId);
+ }
+
+ public void addExecutor(String sessionId, Executor executor) {
+ executors.put(sessionId, executor);
+ }
+}
diff --git
a/paimon-web-server/src/main/java/org/apache/paimon/web/server/service/JobService.java
b/paimon-web-server/src/main/java/org/apache/paimon/web/server/service/JobService.java
new file mode 100644
index 0000000..c7c65f7
--- /dev/null
+++
b/paimon-web-server/src/main/java/org/apache/paimon/web/server/service/JobService.java
@@ -0,0 +1,96 @@
+/*
+ * 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.paimon.web.server.service;
+
+import org.apache.paimon.web.server.data.dto.JobSubmitDTO;
+import org.apache.paimon.web.server.data.dto.ResultFetchDTO;
+import org.apache.paimon.web.server.data.dto.StopJobDTO;
+import org.apache.paimon.web.server.data.model.JobInfo;
+import org.apache.paimon.web.server.data.vo.JobStatisticsVO;
+import org.apache.paimon.web.server.data.vo.JobVO;
+import org.apache.paimon.web.server.data.vo.ResultDataVO;
+
+import com.baomidou.mybatisplus.extension.service.IService;
+
+import java.util.List;
+
+/** Job Service. */
+public interface JobService extends IService<JobInfo> {
+
+ /**
+ * Submits a new job based on the provided job submission data.
+ *
+ * @param jobSubmitDTO the data transfer object containing job submission
details
+ * @return the job view object after submission
+ */
+ JobVO submitJob(JobSubmitDTO jobSubmitDTO);
+
+ /**
+ * Fetches the result of a job specified by the result fetch data transfer
object.
+ *
+ * @param resultFetchDTO the data transfer object containing job result
fetching details
+ * @return result data view object containing the job results
+ */
+ ResultDataVO fetchResult(ResultFetchDTO resultFetchDTO);
+
+ /**
+ * Lists all jobs.
+ *
+ * @return a list of job view objects
+ */
+ List<JobVO> listJobs();
+
+ /**
+ * Lists jobs in a paginated format.
+ *
+ * @param current the current page number
+ * @param size the number of jobs per page
+ * @return a list of job view objects for the specified page
+ */
+ List<JobVO> listJobsByPage(int current, int size);
+
+ /**
+ * Retrieves detailed information about a job identified by its job ID.
+ *
+ * @param id the unique identifier of the job
+ * @return job information object
+ */
+ JobInfo getJobById(String id);
+
+ /**
+ * Retrieves statistics about jobs.
+ *
+ * @return a job statistics view object containing aggregated job data
+ */
+ JobStatisticsVO getJobStatistics();
+
+ /**
+ * Stops a job as specified by the stop job data transfer object.
+ *
+ * @param stopJobDTO the data transfer object containing job stopping
details
+ */
+ void stop(StopJobDTO stopJobDTO);
+
+ /**
+ * Refreshes the status of a job based on the specified task type.
+ *
+ * @param taskType the type of task for which the job status needs to be
updated
+ */
+ void refreshJobStatus(String taskType);
+}
diff --git
a/paimon-web-server/src/main/java/org/apache/paimon/web/server/service/impl/JobServiceImpl.java
b/paimon-web-server/src/main/java/org/apache/paimon/web/server/service/impl/JobServiceImpl.java
new file mode 100644
index 0000000..9c3ffdc
--- /dev/null
+++
b/paimon-web-server/src/main/java/org/apache/paimon/web/server/service/impl/JobServiceImpl.java
@@ -0,0 +1,478 @@
+/*
+ * 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.paimon.web.server.service.impl;
+
+import org.apache.paimon.web.engine.flink.common.executor.Executor;
+import org.apache.paimon.web.engine.flink.common.executor.ExecutorFactory;
+import org.apache.paimon.web.engine.flink.common.result.ExecutionResult;
+import org.apache.paimon.web.engine.flink.common.result.FetchResultParams;
+import org.apache.paimon.web.engine.flink.common.status.JobStatus;
+import org.apache.paimon.web.engine.flink.sql.gateway.model.SessionEntity;
+import org.apache.paimon.web.gateway.config.ExecutionConfig;
+import org.apache.paimon.web.gateway.enums.EngineType;
+import org.apache.paimon.web.gateway.provider.ExecutorFactoryProvider;
+import org.apache.paimon.web.server.data.dto.JobSubmitDTO;
+import org.apache.paimon.web.server.data.dto.ResultFetchDTO;
+import org.apache.paimon.web.server.data.dto.SessionDTO;
+import org.apache.paimon.web.server.data.dto.StopJobDTO;
+import org.apache.paimon.web.server.data.model.ClusterInfo;
+import org.apache.paimon.web.server.data.model.JobInfo;
+import org.apache.paimon.web.server.data.vo.JobStatisticsVO;
+import org.apache.paimon.web.server.data.vo.JobVO;
+import org.apache.paimon.web.server.data.vo.ResultDataVO;
+import org.apache.paimon.web.server.mapper.JobMapper;
+import org.apache.paimon.web.server.service.ClusterService;
+import org.apache.paimon.web.server.service.JobExecutorService;
+import org.apache.paimon.web.server.service.JobService;
+import org.apache.paimon.web.server.service.SessionService;
+import org.apache.paimon.web.server.service.UserSessionManager;
+import org.apache.paimon.web.server.util.LocalDateTimeUtil;
+
+import cn.dev33.satoken.stp.StpUtil;
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import com.baomidou.mybatisplus.core.metadata.IPage;
+import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
+import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.collections.MapUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.scheduling.annotation.Async;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Service;
+
+import java.time.LocalDateTime;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+/** The implementation of {@link JobService}. */
+@Service
+@Slf4j
+public class JobServiceImpl extends ServiceImpl<JobMapper, JobInfo> implements
JobService {
+
+ private static final String STREAMING_MODE = "Streaming";
+ private static final String BATCH_MODE = "Batch";
+ private static final String SHOW_JOBS_STATEMENT = "SHOW JOBS";
+
+ @Autowired private JobMapper jobMapper;
+
+ @Autowired private UserSessionManager sessionManager;
+
+ @Autowired private SessionService sessionService;
+
+ @Autowired private ClusterService clusterService;
+
+ @Autowired private JobExecutorService jobExecutorService;
+
+ @Override
+ public JobVO submitJob(JobSubmitDTO jobSubmitDTO) {
+ String pipelineName = getPipelineName(jobSubmitDTO.getStatements());
+ if (StringUtils.isNotBlank(pipelineName)) {
+ jobSubmitDTO.setJobName(pipelineName);
+ } else {
+ pipelineName = jobSubmitDTO.getJobName();
+ jobSubmitDTO.setStatements(
+ addPipelineNameStatement(pipelineName,
jobSubmitDTO.getStatements()));
+ }
+
+ Executor executor =
+ this.getExecutor(jobSubmitDTO.getClusterId(),
jobSubmitDTO.getTaskType());
+ if (executor == null) {
+ throw new RuntimeException("No executor available for the job
submission.");
+ }
+
+ try {
+ ExecutionResult executionResult =
executor.executeSql(jobSubmitDTO.getStatements());
+ if (StringUtils.isNotBlank(executionResult.getJobId())) {
+ JobInfo jobInfo = buildJobInfo(executionResult, jobSubmitDTO);
+ this.save(jobInfo);
+ }
+ return buildJobVO(executionResult, jobSubmitDTO);
+ } catch (Exception e) {
+ throw new RuntimeException("Error executing job: " +
e.getMessage(), e);
+ }
+ }
+
+ @Override
+ public ResultDataVO fetchResult(ResultFetchDTO resultFetchDTO) {
+ long token = resultFetchDTO.getToken() + 1;
+ try {
+ Executor executor =
+ this.getExecutor(resultFetchDTO.getClusterId(),
resultFetchDTO.getTaskType());
+ if (executor == null) {
+ throw new RuntimeException("No executor available for result
fetching.");
+ }
+ FetchResultParams params =
+ FetchResultParams.builder()
+ .sessionId(resultFetchDTO.getSessionId())
+ .submitId(resultFetchDTO.getSubmitId())
+ .token(token)
+ .build();
+ ExecutionResult executionResult = executor.fetchResults(params);
+ ResultDataVO.ResultDataVOBuilder builder =
+ ResultDataVO.builder()
+ .resultData(executionResult.getData())
+ .rows(executionResult.getData().size())
+ .token(token);
+ if (!executionResult.getData().isEmpty()) {
+
builder.columns(executionResult.getData().get(0).entrySet().size());
+ } else {
+ builder.columns(0);
+ }
+ return builder.build();
+ } catch (Exception e) {
+ throw new RuntimeException("Error fetching result:" +
e.getMessage(), e);
+ }
+ }
+
+ @Override
+ public List<JobVO> listJobs() {
+ List<JobInfo> jobInfos = this.list();
+ return
jobInfos.stream().map(this::convertJobInfoToJobVO).collect(Collectors.toList());
+ }
+
+ @Override
+ public List<JobVO> listJobsByPage(int current, int size) {
+ QueryWrapper<JobInfo> queryWrapper = new QueryWrapper<>();
+ queryWrapper.orderByDesc("create_time");
+ Page<JobInfo> page = new Page<>(current, size);
+ IPage<JobInfo> jobInfoIPage = jobMapper.selectPage(page, queryWrapper);
+ List<JobInfo> records = jobInfoIPage.getRecords();
+ return
records.stream().map(this::convertJobInfoToJobVO).collect(Collectors.toList());
+ }
+
+ @Override
+ public JobInfo getJobById(String id) {
+ QueryWrapper<JobInfo> queryWrapper = new QueryWrapper<>();
+ queryWrapper.eq("job_id", id);
+ return jobMapper.selectOne(queryWrapper);
+ }
+
+ @Override
+ public JobStatisticsVO getJobStatistics() {
+ List<JobInfo> jobInfos = this.list();
+
+ long totalNum = jobInfos.size();
+ long runningNum =
+ jobInfos.stream()
+ .filter(job ->
JobStatus.RUNNING.getValue().equals(job.getStatus()))
+ .count();
+ long finishedNum =
+ jobInfos.stream()
+ .filter(job ->
JobStatus.FINISHED.getValue().equals(job.getStatus()))
+ .count();
+ long canceledNum =
+ jobInfos.stream()
+ .filter(job ->
JobStatus.CANCELED.getValue().equals(job.getStatus()))
+ .count();
+ long failedNum =
+ jobInfos.stream()
+ .filter(job ->
JobStatus.FAILED.getValue().equals(job.getStatus()))
+ .count();
+
+ return JobStatisticsVO.builder()
+ .totalNum(totalNum)
+ .runningNum(runningNum)
+ .finishedNum(finishedNum)
+ .canceledNum(canceledNum)
+ .failedNum(failedNum)
+ .build();
+ }
+
+ @Override
+ public void stop(StopJobDTO stopJobDTO) {
+ try {
+ Executor executor = getExecutor(stopJobDTO.getClusterId(),
stopJobDTO.getTaskType());
+ if (executor == null) {
+ throw new RuntimeException("No executor available for job
stopping.");
+ }
+ executor.stop(stopJobDTO.getJobId(), stopJobDTO.isWithSavepoint());
+ boolean updateStatus =
+ updateJobStatusAndEndTime(
+ stopJobDTO.getJobId(),
+ JobStatus.CANCELED.getValue(),
+ LocalDateTime.now());
+ if (!updateStatus) {
+ log.error(
+ "Failed to update job status in the database for
jobId: {}",
+ stopJobDTO.getJobId());
+ }
+ } catch (Exception e) {
+ throw new RuntimeException("Error stopping job:" + e.getMessage(),
e);
+ }
+ }
+
+ @Override
+ public void refreshJobStatus(String taskType) {
+ if (!StpUtil.isLogin()) {
+ throw new IllegalStateException("User must be logged in to access
this resource");
+ }
+ int userId = StpUtil.getLoginIdAsInt();
+
+ if (taskType.equals("Flink")) {
+ QueryWrapper<ClusterInfo> queryWrapper = new QueryWrapper<>();
+ queryWrapper.eq("type", "Flink");
+ List<ClusterInfo> clusters = clusterService.list();
+ for (ClusterInfo cluster : clusters) {
+ try {
+ SessionEntity session =
+ sessionManager.getSession(userId + "_" +
cluster.getId());
+ if (session == null) {
+ return;
+ }
+
+ Executor executor =
jobExecutorService.getExecutor(session.getSessionId());
+ if (executor == null) {
+ return;
+ }
+
+ ExecutionResult executionResult =
executor.executeSql(SHOW_JOBS_STATEMENT);
+ List<Map<String, Object>> jobsData =
executionResult.getData();
+ for (Map<String, Object> jobData : jobsData) {
+ String jobId = (String) jobData.get("job id");
+ String jobStatus = (String) jobData.get("status");
+ String utcTimeString = (String) jobData.get("start
time");
+ LocalDateTime startTime =
+
LocalDateTimeUtil.convertUtcStringToLocalDateTime(utcTimeString);
+ JobInfo job = getJobById(jobId);
+ if (job != null && job.getUid().equals(userId)) {
+ String currentStatus = job.getStatus();
+ if (!jobStatus.equals(currentStatus)) {
+ if
(JobStatus.RUNNING.getValue().equals(jobStatus)) {
+ updateJobStatusAndStartTime(jobId,
jobStatus, startTime);
+ } else if
(JobStatus.FINISHED.getValue().equals(jobStatus)
+ ||
JobStatus.CANCELED.getValue().equals(jobStatus)) {
+ LocalDateTime endTime =
+ job.getEndTime() == null
+ ? LocalDateTime.now()
+ : job.getEndTime();
+ updateJobStatusAndEndTime(jobId,
jobStatus, endTime);
+ }
+ }
+ } else {
+ log.warn("Job with ID {} not found in the
database.", jobId);
+ }
+ }
+ } catch (Exception e) {
+ log.error("Exception with refreshing job status.", e);
+ }
+ }
+ }
+ }
+
+ @Async
+ @Scheduled(initialDelay = 60000, fixedDelay = 30000)
+ public void initializeExecutorsIfNeeded() {
+ List<SessionEntity> sessions = sessionManager.getAllSessions();
+ for (SessionEntity session : sessions) {
+ try {
+ Executor executor =
jobExecutorService.getExecutor(session.getSessionId());
+ if (executor == null) {
+ ExecutionConfig config =
+
ExecutionConfig.builder().sessionEntity(session).build();
+ EngineType engineType = EngineType.fromName("FLINK");
+ ExecutorFactoryProvider provider = new
ExecutorFactoryProvider(config);
+ ExecutorFactory executorFactory =
provider.getExecutorFactory(engineType);
+ executor = executorFactory.createExecutor();
+ jobExecutorService.addExecutor(session.getSessionId(),
executor);
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(
+ "Failed to create executor for session ID:" +
session.getSessionId(), e);
+ }
+ }
+ }
+
+ private boolean updateJobStatusAndEndTime(
+ String jobId, String newStatus, LocalDateTime endTime) {
+ JobInfo jobInfo = new JobInfo();
+ jobInfo.setStatus(newStatus);
+ jobInfo.setEndTime(endTime);
+ return this.update(jobInfo, new QueryWrapper<JobInfo>().eq("job_id",
jobId));
+ }
+
+ private boolean updateJobStatusAndStartTime(
+ String jobId, String newStatus, LocalDateTime startTime) {
+ JobInfo jobInfo = new JobInfo();
+ jobInfo.setStatus(newStatus);
+ jobInfo.setStartTime(startTime);
+ return this.update(jobInfo, new QueryWrapper<JobInfo>().eq("job_id",
jobId));
+ }
+
+ private JobVO convertJobInfoToJobVO(JobInfo jobInfo) {
+ JobVO.JobVOBuilder builder =
+ JobVO.builder()
+ .jobId(jobInfo.getJobId())
+ .jobName(jobInfo.getJobName())
+ .type(jobInfo.getType())
+ .executeMode(jobInfo.getExecuteMode())
+ .clusterId(jobInfo.getClusterId());
+ if (jobInfo.getStartTime() != null) {
+ builder.startTime(jobInfo.getStartTime());
+ }
+ if (jobInfo.getEndTime() != null) {
+ builder.endTime(jobInfo.getEndTime());
+ }
+ if (StringUtils.isNotBlank(jobInfo.getStatus())) {
+ builder.status(jobInfo.getStatus());
+ }
+ return builder.build();
+ }
+
+ private JobInfo buildJobInfo(ExecutionResult executionResult, JobSubmitDTO
jobSubmitDTO) {
+ JobInfo.JobInfoBuilder builder =
+ JobInfo.builder()
+ .jobId(executionResult.getJobId())
+ .type(jobSubmitDTO.getTaskType())
+ .statements(jobSubmitDTO.getStatements())
+ .status(JobStatus.CREATED.getValue())
+ .clusterId(jobSubmitDTO.getClusterId());
+
+ String jobName = jobSubmitDTO.getJobName() != null ?
jobSubmitDTO.getJobName() : "";
+ builder.jobName(jobName);
+
+ if (StringUtils.isNotBlank(executionResult.getStatus())) {
+ builder.status(executionResult.getStatus());
+ }
+
+ Map<String, String> config =
+ MapUtils.isNotEmpty(jobSubmitDTO.getConfig())
+ ? jobSubmitDTO.getConfig()
+ : new HashMap<>();
+ ObjectMapper objectMapper = new ObjectMapper();
+ String jsonConfig;
+ try {
+ jsonConfig = objectMapper.writeValueAsString(config);
+ } catch (Exception e) {
+ jsonConfig = "{}";
+ }
+ builder.config(jsonConfig);
+
+ String executeMode = jobSubmitDTO.isStreaming() ? STREAMING_MODE :
BATCH_MODE;
+ builder.executeMode(executeMode);
+
+ if (StpUtil.isLogin()) {
+ builder.uid(StpUtil.getLoginIdAsInt());
+ }
+ return builder.build();
+ }
+
+ private JobVO buildJobVO(ExecutionResult executionResult, JobSubmitDTO
jobSubmitDTO) {
+ JobVO.JobVOBuilder builder =
+ JobVO.builder()
+ .type(jobSubmitDTO.getTaskType())
+ .shouldFetchResult(executionResult.shouldFetchResult())
+ .submitId(executionResult.getSubmitId())
+ .resultData(executionResult.getData())
+ .clusterId(jobSubmitDTO.getClusterId())
+ .jobName(jobSubmitDTO.getJobName())
+ .token(0L);
+ String executeMode = jobSubmitDTO.isStreaming() ? STREAMING_MODE :
BATCH_MODE;
+ builder.executeMode(executeMode);
+ if (StringUtils.isNotBlank(executionResult.getJobId())) {
+ builder.jobId(executionResult.getJobId());
+ }
+ if (StringUtils.isNotBlank(executionResult.getStatus())) {
+ builder.status(executionResult.getStatus());
+ }
+ if (StpUtil.isLogin()) {
+ builder.uid(StpUtil.getLoginIdAsInt())
+ .sessionId(
+ sessionManager
+ .getSession(
+ StpUtil.getLoginIdAsInt()
+ + "_"
+ +
jobSubmitDTO.getClusterId())
+ .getSessionId());
+ }
+ return builder.build();
+ }
+
+ private String getPipelineName(String statements) {
+ String regex = "set\\s+'pipeline\\.name'\\s*=\\s*'([^']+)'";
+ Pattern pattern = Pattern.compile(regex, Pattern.CASE_INSENSITIVE);
+ Matcher matcher = pattern.matcher(statements);
+
+ if (matcher.find()) {
+ return matcher.group(1);
+ }
+ return null;
+ }
+
+ private String addPipelineNameStatement(String pipelineName, String
statements) {
+ return "SET 'pipeline.name' = '" + pipelineName + "';\n" + statements;
+ }
+
+ private boolean shouldCreateSession(String clusterId) {
+ if (StpUtil.isLogin()) {
+ SessionEntity session =
+ sessionManager.getSession(StpUtil.getLoginIdAsInt() + "_"
+ clusterId);
+ if (session != null) {
+ SessionDTO sessionDTO = new SessionDTO();
+ sessionDTO.setHost(session.getHost());
+ sessionDTO.setPort(session.getPort());
+ sessionDTO.setUid(StpUtil.getLoginIdAsInt());
+ sessionDTO.setClusterId(Integer.valueOf(clusterId));
+ return sessionService.triggerSessionHeartbeat(sessionDTO) <= 0;
+ }
+ }
+ return true;
+ }
+
+ private Executor getExecutor(String clusterId, String taskType) {
+ try {
+ if (!StpUtil.isLogin()) {
+ throw new IllegalStateException("User must be logged in to
access this resource");
+ }
+ if (shouldCreateSession(clusterId)) {
+ ClusterInfo clusterInfo = clusterService.getById(clusterId);
+ if (clusterInfo == null) {
+ throw new IllegalStateException("No cluster found with ID:
" + clusterId);
+ }
+ SessionDTO sessionDTO = new SessionDTO();
+ sessionDTO.setHost(clusterInfo.getHost());
+ sessionDTO.setPort(clusterInfo.getPort());
+ sessionDTO.setUid(StpUtil.getLoginIdAsInt());
+ sessionDTO.setClusterId(Integer.valueOf(clusterId));
+ sessionService.createSession(sessionDTO);
+ }
+
+ SessionEntity session =
+ sessionManager.getSession(StpUtil.getLoginIdAsInt() + "_"
+ clusterId);
+
+ if (jobExecutorService.getExecutor(session.getSessionId()) ==
null) {
+ ExecutionConfig config =
ExecutionConfig.builder().sessionEntity(session).build();
+ EngineType engineType =
EngineType.fromName(taskType.toUpperCase());
+ ExecutorFactoryProvider provider = new
ExecutorFactoryProvider(config);
+ ExecutorFactory executorFactory =
provider.getExecutorFactory(engineType);
+ Executor executor = executorFactory.createExecutor();
+ jobExecutorService.addExecutor(session.getSessionId(),
executor);
+ }
+ return jobExecutorService.getExecutor(session.getSessionId());
+ } catch (Exception e) {
+ log.error("Failed to create executor: {}", e.getMessage(), e);
+ }
+ return null;
+ }
+}
diff --git
a/paimon-web-server/src/main/java/org/apache/paimon/web/server/util/LocalDateTimeUtil.java
b/paimon-web-server/src/main/java/org/apache/paimon/web/server/util/LocalDateTimeUtil.java
new file mode 100644
index 0000000..7443f97
--- /dev/null
+++
b/paimon-web-server/src/main/java/org/apache/paimon/web/server/util/LocalDateTimeUtil.java
@@ -0,0 +1,36 @@
+/*
+ * 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.paimon.web.server.util;
+
+import java.time.LocalDateTime;
+import java.time.OffsetDateTime;
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
+import java.time.format.DateTimeFormatter;
+
+/** Local date time util. */
+public class LocalDateTimeUtil {
+
+ public static LocalDateTime convertUtcStringToLocalDateTime(String
utcTimeStr) {
+ OffsetDateTime utcTime =
+ OffsetDateTime.parse(utcTimeStr + "Z",
DateTimeFormatter.ISO_OFFSET_DATE_TIME);
+ ZonedDateTime beijingTime =
utcTime.atZoneSameInstant(ZoneId.of("Asia/Shanghai"));
+ return beijingTime.toLocalDateTime();
+ }
+}
diff --git a/paimon-web-server/src/main/resources/i18n/messages.properties
b/paimon-web-server/src/main/resources/i18n/messages.properties
index 96ae9d1..546bf8d 100644
--- a/paimon-web-server/src/main/resources/i18n/messages.properties
+++ b/paimon-web-server/src/main/resources/i18n/messages.properties
@@ -53,3 +53,7 @@ cdc.job.exist.error=Paimon CDC job exists.
cdc.job.not.exist.error=Paimon CDC job is not exist.
cluster.not.exist=This cluster is not exist.
cluster.name.exist=This cluster name {0} already exists.
+job.submit.error=Exception submitting a job.
+result.fetch.error=Exception fetching result data.
+job.stop.error=Exception stopping a job.
+job.update.status.error=Exception updating the job status.
diff --git
a/paimon-web-server/src/main/resources/i18n/messages_en_US.properties
b/paimon-web-server/src/main/resources/i18n/messages_en_US.properties
index 96ae9d1..546bf8d 100644
--- a/paimon-web-server/src/main/resources/i18n/messages_en_US.properties
+++ b/paimon-web-server/src/main/resources/i18n/messages_en_US.properties
@@ -53,3 +53,7 @@ cdc.job.exist.error=Paimon CDC job exists.
cdc.job.not.exist.error=Paimon CDC job is not exist.
cluster.not.exist=This cluster is not exist.
cluster.name.exist=This cluster name {0} already exists.
+job.submit.error=Exception submitting a job.
+result.fetch.error=Exception fetching result data.
+job.stop.error=Exception stopping a job.
+job.update.status.error=Exception updating the job status.
diff --git
a/paimon-web-server/src/main/resources/i18n/messages_zh_CN.properties
b/paimon-web-server/src/main/resources/i18n/messages_zh_CN.properties
index d61de0c..3ea53d8 100644
--- a/paimon-web-server/src/main/resources/i18n/messages_zh_CN.properties
+++ b/paimon-web-server/src/main/resources/i18n/messages_zh_CN.properties
@@ -53,3 +53,7 @@ cdc.job.exist.error=paimon cdc\u4F5C\u4E1A\u5DF2\u5B58\u5728
cdc.job.not.exist.error=paimon cdc\u4F5C\u4E1A\u4E0D\u5B58\u5728
cluster.not.exist=\u6B64\u96C6\u7FA4\u4E0D\u5B58\u5728
cluster.name.exist=\u6B64\u96C6\u7FA4\u540D{0}\u5DF2\u7ECF\u5B58\u5728
+job.submit.error=\u63D0\u4EA4\u4F5C\u4E1A\u65F6\u53D1\u751F\u5F02\u5E38
+result.fetch.error=\u83B7\u53D6\u7ED3\u679C\u6570\u636E\u65F6\u53D1\u751F\u5F02\u5E38
+job.stop.error=\u505C\u6B62\u4F5C\u4E1A\u65F6\u53D1\u751F\u5F02\u5E38
+job.update.status.error=\u66F4\u65B0\u4F5C\u4E1A\u72B6\u6001\u65F6\u53D1\u751F\u5F02\u5E38
diff --git
a/paimon-web-server/src/test/java/org/apache/paimon/web/server/constant/StatementsConstant.java
b/paimon-web-server/src/test/java/org/apache/paimon/web/server/constant/StatementsConstant.java
new file mode 100644
index 0000000..7d7ce3f
--- /dev/null
+++
b/paimon-web-server/src/test/java/org/apache/paimon/web/server/constant/StatementsConstant.java
@@ -0,0 +1,76 @@
+/*
+ * 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.paimon.web.server.constant;
+
+/** Statements constant. */
+public class StatementsConstant {
+
+ public static String statement =
+ "CREATE TABLE IF NOT EXISTS t_order(\n"
+ + " `order_id` BIGINT,\n"
+ + " `product` BIGINT,\n"
+ + " `amount` BIGINT,\n"
+ + " `order_time` as CAST(CURRENT_TIMESTAMP AS
TIMESTAMP(3)),\n"
+ + " WATERMARK FOR order_time AS order_time-INTERVAL '2'
SECOND\n"
+ + ") WITH(\n"
+ + " 'connector' = 'datagen',\n"
+ + " 'rows-per-second' = '1',\n"
+ + " 'fields.order_id.min' = '1',\n"
+ + " 'fields.order_id.max' = '2',\n"
+ + " 'fields.amount.min' = '1',\n"
+ + " 'fields.amount.max' = '10',\n"
+ + " 'fields.product.min' = '1',\n"
+ + " 'fields.product.max' = '2'\n"
+ + ");\n"
+ + "CREATE TABLE IF NOT EXISTS sink_table(\n"
+ + " `product` BIGINT,\n"
+ + " `amount` BIGINT,\n"
+ + " `order_time` TIMESTAMP(3),\n"
+ + " `one_minute_sum` BIGINT\n"
+ + ") WITH('connector' = 'print');\n"
+ + "\n"
+ + "INSERT INTO\n"
+ + " sink_table\n"
+ + "SELECT\n"
+ + " product,\n"
+ + " amount,\n"
+ + " order_time,\n"
+ + " 0 as one_minute_sum\n"
+ + "FROM\n"
+ + " t_order;";
+
+ public static String selectStatement =
+ "CREATE TABLE IF NOT EXISTS t_order(\n"
+ + " `order_id` BIGINT,\n"
+ + " `product` BIGINT,\n"
+ + " `amount` BIGINT,\n"
+ + " `order_time` as CAST(CURRENT_TIMESTAMP AS
TIMESTAMP(3)),\n"
+ + " WATERMARK FOR order_time AS order_time-INTERVAL '2'
SECOND\n"
+ + ") WITH(\n"
+ + " 'connector' = 'datagen',\n"
+ + " 'rows-per-second' = '1',\n"
+ + " 'fields.order_id.min' = '1',\n"
+ + " 'fields.order_id.max' = '2',\n"
+ + " 'fields.amount.min' = '1',\n"
+ + " 'fields.amount.max' = '10',\n"
+ + " 'fields.product.min' = '1',\n"
+ + " 'fields.product.max' = '2'\n"
+ + ");\n"
+ + "SELECT * FROM t_order;";
+}
diff --git
a/paimon-web-server/src/test/java/org/apache/paimon/web/server/controller/JobControllerTest.java
b/paimon-web-server/src/test/java/org/apache/paimon/web/server/controller/JobControllerTest.java
new file mode 100644
index 0000000..bac3f46
--- /dev/null
+++
b/paimon-web-server/src/test/java/org/apache/paimon/web/server/controller/JobControllerTest.java
@@ -0,0 +1,404 @@
+/*
+ * 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.paimon.web.server.controller;
+
+import org.apache.paimon.web.server.constant.StatementsConstant;
+import org.apache.paimon.web.server.data.dto.JobSubmitDTO;
+import org.apache.paimon.web.server.data.dto.LoginDTO;
+import org.apache.paimon.web.server.data.dto.ResultFetchDTO;
+import org.apache.paimon.web.server.data.dto.StopJobDTO;
+import org.apache.paimon.web.server.data.model.ClusterInfo;
+import org.apache.paimon.web.server.data.result.R;
+import org.apache.paimon.web.server.data.vo.JobStatisticsVO;
+import org.apache.paimon.web.server.data.vo.JobStatusVO;
+import org.apache.paimon.web.server.data.vo.JobVO;
+import org.apache.paimon.web.server.data.vo.ResultDataVO;
+import org.apache.paimon.web.server.service.ClusterService;
+import org.apache.paimon.web.server.util.ObjectMapperUtils;
+import org.apache.paimon.web.server.util.StringUtils;
+
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import com.fasterxml.jackson.core.type.TypeReference;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.MethodOrderer;
+import org.junit.jupiter.api.Order;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestMethodOrder;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import
org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.http.MediaType;
+import org.springframework.mock.web.MockCookie;
+import org.springframework.mock.web.MockHttpServletResponse;
+import org.springframework.test.web.servlet.MockMvc;
+import org.springframework.test.web.servlet.request.MockMvcRequestBuilders;
+import org.springframework.test.web.servlet.result.MockMvcResultHandlers;
+import org.springframework.test.web.servlet.result.MockMvcResultMatchers;
+
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/** Tests for {@link JobController}. */
+@SpringBootTest
+@AutoConfigureMockMvc
+@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
+public class JobControllerTest extends FlinkSQLGatewayTestBase {
+
+ private static final String loginPath = "/api/login";
+ private static final String logoutPath = "/api/logout";
+ private static final String jobPath = "/api/job";
+
+ @Value("${spring.application.name}")
+ private String tokenName;
+
+ @Autowired public MockMvc mockMvc;
+
+ public static MockCookie cookie;
+
+ @Autowired private ClusterService clusterService;
+
+ @BeforeEach
+ public void before() throws Exception {
+ LoginDTO login = new LoginDTO();
+ login.setUsername("admin");
+ login.setPassword("admin");
+ MockHttpServletResponse response =
+ mockMvc.perform(
+ MockMvcRequestBuilders.post(loginPath)
+
.content(ObjectMapperUtils.toJSON(login))
+
.contentType(MediaType.APPLICATION_JSON_VALUE)
+
.accept(MediaType.APPLICATION_JSON_VALUE))
+ .andExpect(MockMvcResultMatchers.status().isOk())
+ .andDo(MockMvcResultHandlers.print())
+ .andReturn()
+ .getResponse();
+ String result = response.getContentAsString();
+ R<?> r = ObjectMapperUtils.fromJSON(result, R.class);
+ assertEquals(200, r.getCode());
+
+ assertTrue(StringUtils.isNotBlank(r.getData().toString()));
+
+ cookie = (MockCookie) response.getCookie(tokenName);
+
+ ClusterInfo cluster =
+ ClusterInfo.builder()
+ .clusterName("test_cluster")
+ .host(targetAddress)
+ .port(port)
+ .enabled(true)
+ .type("Flink")
+ .build();
+ boolean res = clusterService.save(cluster);
+ assertTrue(res);
+ }
+
+ @AfterEach
+ public void after() throws Exception {
+ String result =
+ mockMvc.perform(
+ MockMvcRequestBuilders.post(logoutPath)
+ .cookie(cookie)
+
.contentType(MediaType.APPLICATION_JSON_VALUE)
+
.accept(MediaType.APPLICATION_JSON_VALUE))
+ .andExpect(MockMvcResultMatchers.status().isOk())
+ .andDo(MockMvcResultHandlers.print())
+ .andReturn()
+ .getResponse()
+ .getContentAsString();
+ R<?> r = ObjectMapperUtils.fromJSON(result, R.class);
+ assertEquals(200, r.getCode());
+
+ QueryWrapper<ClusterInfo> queryWrapper = new QueryWrapper<>();
+ queryWrapper.eq("cluster_name", "test_cluster");
+ ClusterInfo one = clusterService.getOne(queryWrapper);
+ int res = clusterService.deleteClusterByIds(new Integer[]
{one.getId()});
+ assertTrue(res > 0);
+ }
+
+ @Test
+ @Order(1)
+ public void testSubmitJob() throws Exception {
+ QueryWrapper<ClusterInfo> queryWrapper = new QueryWrapper<>();
+ queryWrapper.eq("cluster_name", "test_cluster");
+ ClusterInfo one = clusterService.getOne(queryWrapper);
+ JobSubmitDTO jobSubmitDTO = new JobSubmitDTO();
+ jobSubmitDTO.setJobName("flink-job-test");
+ jobSubmitDTO.setTaskType("Flink");
+ jobSubmitDTO.setClusterId(String.valueOf(one.getId()));
+ jobSubmitDTO.setStatements(StatementsConstant.statement);
+
+ String responseString = submit(jobSubmitDTO);
+
+ R<JobVO> r = ObjectMapperUtils.fromJSON(responseString, new
TypeReference<R<JobVO>>() {});
+ assertEquals(200, r.getCode());
+ assertNotNull(r.getData());
+ assertEquals("flink-job-test", r.getData().getJobName());
+ assertEquals("Flink", r.getData().getType());
+ assertEquals(1, r.getData().getUid());
+ }
+
+ @Test
+ @Order(2)
+ public void testFetchResult() throws Exception {
+ QueryWrapper<ClusterInfo> queryWrapper = new QueryWrapper<>();
+ queryWrapper.eq("cluster_name", "test_cluster");
+ ClusterInfo one = clusterService.getOne(queryWrapper);
+ JobSubmitDTO jobSubmitDTO = new JobSubmitDTO();
+ jobSubmitDTO.setJobName("flink-job-test-fetch-result");
+ jobSubmitDTO.setTaskType("Flink");
+ jobSubmitDTO.setClusterId(String.valueOf(one.getId()));
+ jobSubmitDTO.setStatements(StatementsConstant.selectStatement);
+
+ String responseString = submit(jobSubmitDTO);
+ R<JobVO> r = ObjectMapperUtils.fromJSON(responseString, new
TypeReference<R<JobVO>>() {});
+ assertEquals(200, r.getCode());
+
+ if (r.getData().getShouldFetchResult()) {
+ ResultFetchDTO resultFetchDTO = new ResultFetchDTO();
+ resultFetchDTO.setSubmitId(r.getData().getSubmitId());
+ resultFetchDTO.setClusterId(r.getData().getClusterId());
+ resultFetchDTO.setSessionId(r.getData().getSessionId());
+ resultFetchDTO.setTaskType(r.getData().getType());
+ resultFetchDTO.setToken(r.getData().getToken());
+
+ String fetchResultString =
+ mockMvc.perform(
+ MockMvcRequestBuilders.post(jobPath +
"/fetch")
+ .cookie(cookie)
+
.content(ObjectMapperUtils.toJSON(resultFetchDTO))
+
.contentType(MediaType.APPLICATION_JSON_VALUE)
+
.accept(MediaType.APPLICATION_JSON_VALUE))
+ .andExpect(MockMvcResultMatchers.status().isOk())
+ .andDo(MockMvcResultHandlers.print())
+ .andReturn()
+ .getResponse()
+ .getContentAsString();
+ R<ResultDataVO> fetchResult =
+ ObjectMapperUtils.fromJSON(
+ fetchResultString, new
TypeReference<R<ResultDataVO>>() {});
+ assertEquals(200, fetchResult.getCode());
+ assertEquals(1, fetchResult.getData().getToken());
+ }
+ }
+
+ @Test
+ @Order(3)
+ public void testListJobs() throws Exception {
+ QueryWrapper<ClusterInfo> queryWrapper = new QueryWrapper<>();
+ queryWrapper.eq("cluster_name", "test_cluster");
+ ClusterInfo one = clusterService.getOne(queryWrapper);
+ JobSubmitDTO jobSubmitDTO = new JobSubmitDTO();
+ jobSubmitDTO.setJobName("flink-job-test-list-jobs");
+ jobSubmitDTO.setTaskType("Flink");
+ jobSubmitDTO.setClusterId(String.valueOf(one.getId()));
+ jobSubmitDTO.setStatements(StatementsConstant.selectStatement);
+
+ String responseString = submit(jobSubmitDTO);
+ R<JobVO> r = ObjectMapperUtils.fromJSON(responseString, new
TypeReference<R<JobVO>>() {});
+ assertEquals(200, r.getCode());
+
+ String listJobResponseStr =
+ mockMvc.perform(
+ MockMvcRequestBuilders.get(jobPath + "/list")
+ .cookie(cookie)
+
.contentType(MediaType.APPLICATION_JSON_VALUE)
+
.accept(MediaType.APPLICATION_JSON_VALUE))
+ .andExpect(MockMvcResultMatchers.status().isOk())
+ .andDo(MockMvcResultHandlers.print())
+ .andReturn()
+ .getResponse()
+ .getContentAsString();
+ R<List<JobVO>> listJobRes =
+ ObjectMapperUtils.fromJSON(
+ listJobResponseStr, new
TypeReference<R<List<JobVO>>>() {});
+ assertEquals(200, listJobRes.getCode());
+ assertEquals(3, listJobRes.getData().size());
+ String jobs =
+ listJobRes.getData().get(0).getJobName()
+ + ","
+ + listJobRes.getData().get(1).getJobName()
+ + ","
+ + listJobRes.getData().get(2).getJobName();
+ assertTrue(jobs.contains("flink-job-test-list-jobs"));
+ }
+
+ @Test
+ @Order(4)
+ public void testGetJobStatus() throws Exception {
+ QueryWrapper<ClusterInfo> queryWrapper = new QueryWrapper<>();
+ queryWrapper.eq("cluster_name", "test_cluster");
+ ClusterInfo one = clusterService.getOne(queryWrapper);
+ JobSubmitDTO jobSubmitDTO = new JobSubmitDTO();
+ jobSubmitDTO.setJobName("flink-job-test-get-job-status");
+ jobSubmitDTO.setTaskType("Flink");
+ jobSubmitDTO.setClusterId(String.valueOf(one.getId()));
+ jobSubmitDTO.setStatements(StatementsConstant.selectStatement);
+
+ String responseString = submit(jobSubmitDTO);
+ R<JobVO> r = ObjectMapperUtils.fromJSON(responseString, new
TypeReference<R<JobVO>>() {});
+ assertEquals(200, r.getCode());
+
+ String listJobResponseStr = getJobStatus(r.getData().getJobId());
+ R<JobStatusVO> getJobStatusRes =
+ ObjectMapperUtils.fromJSON(
+ listJobResponseStr, new
TypeReference<R<JobStatusVO>>() {});
+ assertEquals(200, getJobStatusRes.getCode());
+ assertEquals(r.getData().getJobId(),
getJobStatusRes.getData().getJobId());
+ assertNotNull(getJobStatusRes.getData().getStatus());
+ }
+
+ @Test
+ @Order(5)
+ public void testStopJob() throws Exception {
+ QueryWrapper<ClusterInfo> queryWrapper = new QueryWrapper<>();
+ queryWrapper.eq("cluster_name", "test_cluster");
+ ClusterInfo one = clusterService.getOne(queryWrapper);
+ JobSubmitDTO jobSubmitDTO = new JobSubmitDTO();
+ jobSubmitDTO.setJobName("flink-job-test-stop-job");
+ jobSubmitDTO.setTaskType("Flink");
+ jobSubmitDTO.setClusterId(String.valueOf(one.getId()));
+ jobSubmitDTO.setStatements(StatementsConstant.selectStatement);
+ String responseString = submit(jobSubmitDTO);
+ R<JobVO> r = ObjectMapperUtils.fromJSON(responseString, new
TypeReference<R<JobVO>>() {});
+ assertEquals(200, r.getCode());
+
+ String jobStatus = getJobStatus(r.getData().getJobId());
+ R<JobStatusVO> getJobStatusRes =
+ ObjectMapperUtils.fromJSON(jobStatus, new
TypeReference<R<JobStatusVO>>() {});
+ while (!getJobStatusRes.getData().getStatus().equals("RUNNING")) {
+ refreshJob();
+ jobStatus = getJobStatus(r.getData().getJobId());
+ getJobStatusRes =
+ ObjectMapperUtils.fromJSON(jobStatus, new
TypeReference<R<JobStatusVO>>() {});
+ TimeUnit.SECONDS.sleep(1);
+ }
+
+ StopJobDTO stopJobDTO = new StopJobDTO();
+ stopJobDTO.setJobId(r.getData().getJobId());
+ stopJobDTO.setClusterId(r.getData().getClusterId());
+ stopJobDTO.setTaskType(r.getData().getType());
+ String stopJobString =
+ mockMvc.perform(
+ MockMvcRequestBuilders.post(jobPath + "/stop")
+ .cookie(cookie)
+
.content(ObjectMapperUtils.toJSON(stopJobDTO))
+
.contentType(MediaType.APPLICATION_JSON_VALUE)
+
.accept(MediaType.APPLICATION_JSON_VALUE))
+ .andExpect(MockMvcResultMatchers.status().isOk())
+ .andDo(MockMvcResultHandlers.print())
+ .andReturn()
+ .getResponse()
+ .getContentAsString();
+ R<Void> stopJobRes =
+ ObjectMapperUtils.fromJSON(stopJobString, new
TypeReference<R<Void>>() {});
+ assertEquals(200, stopJobRes.getCode());
+
+ jobStatus = getJobStatus(r.getData().getJobId());
+ getJobStatusRes =
+ ObjectMapperUtils.fromJSON(jobStatus, new
TypeReference<R<JobStatusVO>>() {});
+ assertEquals(200, getJobStatusRes.getCode());
+ assertEquals("CANCELED", getJobStatusRes.getData().getStatus());
+ }
+
+ @Test
+ @Order(6)
+ public void testGetJobStatistics() throws Exception {
+ QueryWrapper<ClusterInfo> queryWrapper = new QueryWrapper<>();
+ queryWrapper.eq("cluster_name", "test_cluster");
+ ClusterInfo one = clusterService.getOne(queryWrapper);
+ JobSubmitDTO jobSubmitDTO = new JobSubmitDTO();
+ jobSubmitDTO.setJobName("flink-job-test-get-job-statistics");
+ jobSubmitDTO.setTaskType("Flink");
+ jobSubmitDTO.setClusterId(String.valueOf(one.getId()));
+ jobSubmitDTO.setStatements(StatementsConstant.selectStatement);
+ String responseString = submit(jobSubmitDTO);
+ R<JobVO> r = ObjectMapperUtils.fromJSON(responseString, new
TypeReference<R<JobVO>>() {});
+ assertEquals(200, r.getCode());
+ refreshJob();
+ TimeUnit.MICROSECONDS.sleep(100);
+ refreshJob();
+
+ String getJobStatisticsResponseStr =
+ mockMvc.perform(
+ MockMvcRequestBuilders.get(jobPath +
"/statistics/get")
+ .cookie(cookie)
+
.contentType(MediaType.APPLICATION_JSON_VALUE)
+
.accept(MediaType.APPLICATION_JSON_VALUE))
+ .andExpect(MockMvcResultMatchers.status().isOk())
+ .andDo(MockMvcResultHandlers.print())
+ .andReturn()
+ .getResponse()
+ .getContentAsString();
+ R<JobStatisticsVO> getJobStatisticsRes =
+ ObjectMapperUtils.fromJSON(
+ getJobStatisticsResponseStr, new
TypeReference<R<JobStatisticsVO>>() {});
+ assertEquals(200, getJobStatisticsRes.getCode());
+ assertEquals(6, getJobStatisticsRes.getData().getTotalNum());
+ assertEquals(5, getJobStatisticsRes.getData().getRunningNum());
+ assertEquals(1, getJobStatisticsRes.getData().getCanceledNum());
+ assertEquals(0, getJobStatisticsRes.getData().getFinishedNum());
+ assertEquals(0, getJobStatisticsRes.getData().getFailedNum());
+ }
+
+ private String submit(JobSubmitDTO jobSubmitDTO) throws Exception {
+ return mockMvc.perform(
+ MockMvcRequestBuilders.post(jobPath + "/submit")
+ .cookie(cookie)
+
.content(ObjectMapperUtils.toJSON(jobSubmitDTO))
+ .contentType(MediaType.APPLICATION_JSON_VALUE)
+ .accept(MediaType.APPLICATION_JSON_VALUE))
+ .andExpect(MockMvcResultMatchers.status().isOk())
+ .andDo(MockMvcResultHandlers.print())
+ .andReturn()
+ .getResponse()
+ .getContentAsString();
+ }
+
+ private String getJobStatus(String jobId) throws Exception {
+ return mockMvc.perform(
+ MockMvcRequestBuilders.get(jobPath + "/status/get/" +
jobId)
+ .cookie(cookie)
+ .contentType(MediaType.APPLICATION_JSON_VALUE)
+ .accept(MediaType.APPLICATION_JSON_VALUE))
+ .andExpect(MockMvcResultMatchers.status().isOk())
+ .andDo(MockMvcResultHandlers.print())
+ .andReturn()
+ .getResponse()
+ .getContentAsString();
+ }
+
+ private void refreshJob() throws Exception {
+ mockMvc.perform(
+ MockMvcRequestBuilders.post(jobPath + "/refresh")
+ .cookie(cookie)
+ .contentType(MediaType.APPLICATION_JSON_VALUE)
+ .accept(MediaType.APPLICATION_JSON_VALUE))
+ .andExpect(MockMvcResultMatchers.status().isOk())
+ .andDo(MockMvcResultHandlers.print())
+ .andReturn()
+ .getResponse()
+ .getContentAsString();
+ }
+}
diff --git a/scripts/sql/paimon-mysql.sql b/scripts/sql/paimon-mysql.sql
index 91f234f..5cc084e 100644
--- a/scripts/sql/paimon-mysql.sql
+++ b/scripts/sql/paimon-mysql.sql
@@ -155,6 +155,24 @@ CREATE TABLE if not exists `cdc_job_definition`
unique (`name`)
) engine = innodb;
+CREATE TABLE if not exists `job`
+(
+ `id` int(11) not null auto_increment primary key comment 'id',
+ `job_id` varchar(100) not null comment 'job id',
+ `job_name` varchar(200) comment 'job name',
+ `type` varchar(100) comment 'job type',
+ `execute_mode` varchar(50) comment 'execute mode',
+ `cluster_id` varchar(100) comment 'cluster id',
+ `uid` INT(11) COMMENT 'User ID',
+ `config` text comment 'config',
+ `statements` text COMMENT 'statements',
+ `status` varchar(50) COMMENT 'status',
+ `start_time` datetime(0) NULL COMMENT 'start time',
+ `end_time` datetime(0) NULL COMMENT 'end time',
+ `create_time` datetime(0) NULL DEFAULT CURRENT_TIMESTAMP COMMENT 'create
time',
+ `update_time` datetime(0) NULL DEFAULT CURRENT_TIMESTAMP COMMENT 'update
time'
+ ) ENGINE = InnoDB DEFAULT CHARSET=utf8;
+
INSERT INTO `user` ( id, username, password, nickname, mobile
, email, enabled, is_delete)
VALUES ( 1, 'admin', '21232f297a57a5a743894a0e4a801fc3', 'Admin', 0
