[Improve][RestAPI] return finished job info when job is finished (#6576)

Author: liunaijie

Diff for: docs/en/seatunnel-engine/rest-api.md

@@ -92,25 +92,31 @@ network:
   "jobId": "",
   "jobName": "",
   "jobStatus": "",
-  "envOptions": {
-  },
   "createTime": "",
   "jobDag": {
     "vertices": [
     ],
     "edges": [
     ]
   },
-  "pluginJarsUrls": [
-  ],
-  "isStartWithSavePoint": false,
   "metrics": {
     "sourceReceivedCount": "",
     "sinkWriteCount": ""
-  }
+  },
+  "finishedTime": "",
+  "errorMsg": null,
+  "envOptions": {
+  },
+  "pluginJarsUrls": [
+  ],
+  "isStartWithSavePoint": false
 }
 ```
 
+`jobId`, `jobName`, `jobStatus`, `createTime`, `jobDag`, `metrics` always be returned.
+`envOptions`, `pluginJarsUrls`, `isStartWithSavePoint` will return when job is running.
+`finishedTime`, `errorMsg` will return when job is finished.
+
 When we can't get the job info, the response will be:
 
 ```json

Diff for: seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/RestApiIT.java

@@ -52,6 +52,8 @@ public class RestApiIT {
 
     private static ClientJobProxy clientJobProxy;
 
+    private static ClientJobProxy batchJobProxy;
+
     private static HazelcastInstanceImpl node1;
 
     private static HazelcastInstanceImpl node2;
@@ -85,6 +87,19 @@ void beforeClass() throws Exception {
                         () ->
                                 Assertions.assertEquals(
                                         JobStatus.RUNNING, clientJobProxy.getJobStatus()));
+
+        String batchFilePath = TestUtils.getResource("fakesource_to_console.conf");
+        JobConfig batchConf = new JobConfig();
+        batchConf.setName("fake_to_console");
+        ClientJobExecutionEnvironment batchJobExecutionEnv =
+                engineClient.createExecutionContext(batchFilePath, batchConf, seaTunnelConfig);
+        batchJobProxy = batchJobExecutionEnv.execute();
+        Awaitility.await()
+                .atMost(5, TimeUnit.MINUTES)
+                .untilAsserted(
+                        () ->
+                                Assertions.assertEquals(
+                                        JobStatus.FINISHED, batchJobProxy.getJobStatus()));
     }
 
     @Test
@@ -108,6 +123,27 @@ public void testGetRunningJobById() {
                         });
     }
 
+    @Test
+    public void testGetJobById() {
+        Arrays.asList(node2, node1)
+                .forEach(
+                        instance -> {
+                            given().get(
+                                            HOST
+                                                    + instance.getCluster()
+                                                            .getLocalMember()
+                                                            .getAddress()
+                                                            .getPort()
+                                                    + RestConstant.RUNNING_JOB_URL
+                                                    + "/"
+                                                    + batchJobProxy.getJobId())
+                                    .then()
+                                    .statusCode(200)
+                                    .body("jobName", equalTo("fake_to_console"))
+                                    .body("jobStatus", equalTo("FINISHED"));
+                        });
+    }
+
     @Test
     public void testGetAnNotExistJobById() {
         Arrays.asList(node2, node1)

Diff for: seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/RestHttpGetCommandProcessor.java

@@ -272,8 +272,36 @@ private void handleJobInfoById(HttpGetCommand command, String uri) {
                                 .getMap(Constant.IMAP_RUNNING_JOB_INFO)
                                 .get(Long.valueOf(jobId));
 
+        JobState finishedJobState =
+                (JobState)
+                        this.textCommandService
+                                .getNode()
+                                .getNodeEngine()
+                                .getHazelcastInstance()
+                                .getMap(Constant.IMAP_FINISHED_JOB_STATE)
+                                .get(Long.valueOf(jobId));
         if (!jobId.isEmpty() && jobInfo != null) {
             this.prepareResponse(command, convertToJson(jobInfo, Long.parseLong(jobId)));
+        } else if (!jobId.isEmpty() && finishedJobState != null) {
+            JobMetrics finishedJobMetrics =
+                    (JobMetrics)
+                            this.textCommandService
+                                    .getNode()
+                                    .getNodeEngine()
+                                    .getHazelcastInstance()
+                                    .getMap(Constant.IMAP_FINISHED_JOB_METRICS)
+                                    .get(Long.valueOf(jobId));
+            JobDAGInfo finishedJobDAGInfo =
+                    (JobDAGInfo)
+                            this.textCommandService
+                                    .getNode()
+                                    .getNodeEngine()
+                                    .getHazelcastInstance()
+                                    .getMap(Constant.IMAP_FINISHED_JOB_VERTEX_INFO)
+                                    .get(Long.valueOf(jobId));
+            this.prepareResponse(
+                    command,
+                    convertToJson(finishedJobState, finishedJobMetrics, finishedJobDAGInfo));
         } else {
             this.prepareResponse(command, new JsonObject().add(RestConstant.JOB_ID, jobId));
         }
@@ -411,6 +439,34 @@ private JsonObject convertToJson(JobInfo jobInfo, long jobId) {
         return jobInfoJson;
     }
 
+    private JsonObject convertToJson(
+            JobState finishedJobState,
+            JobMetrics finishedJobMetrics,
+            JobDAGInfo finishedJobDAGInfo) {
+        JsonObject jobInfoJson = new JsonObject();
+        jobInfoJson
+                .add(RestConstant.JOB_ID, String.valueOf(finishedJobState.getJobId()))
+                .add(RestConstant.JOB_NAME, finishedJobState.getJobName())
+                .add(RestConstant.JOB_STATUS, finishedJobState.getJobStatus().toString())
+                .add(RestConstant.ERROR_MSG, finishedJobState.getErrorMessage())
+                .add(
+                        RestConstant.CREATE_TIME,
+                        new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
+                                .format(new Date(finishedJobState.getSubmitTime())))
+                .add(
+                        RestConstant.FINISH_TIME,
+                        new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
+                                .format(new Date(finishedJobState.getFinishTime())))
+                .add(
+                        RestConstant.JOB_DAG,
+                        Json.parse(JsonUtils.toJsonString(finishedJobDAGInfo)).asObject())
+                .add(RestConstant.PLUGIN_JARS_URLS, new JsonArray())
+                .add(
+                        RestConstant.METRICS,
+                        JsonUtil.toJsonObject(getJobMetrics(finishedJobMetrics.toJsonString())));
+        return jobInfoJson;
+    }
+
     private JsonObject convertToJson(
             JobState jobState, String jobMetrics, JsonObject jobDAGInfo, long jobId) {
         JsonObject jobInfoJson = new JsonObject();