[FLINK-38760][runtime] Add REST APIs for application query and cancellation

flink-runtime-web/src/test/resources/rest_api_v1.snapshot

@@ -1,5 +1,167 @@
 {
   "calls" : [ {
+    "url" : "/applications/overview",
+    "method" : "GET",
+    "status-code" : "200 OK",
+    "file-upload" : false,
+    "path-parameters" : {
+      "pathParameters" : [ ]
+    },
+    "query-parameters" : {
+      "queryParameters" : [ ]
+    },
+    "request" : {
+      "type" : "object",
+      "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:EmptyRequestBody"
+    },
+    "response" : {
+      "type" : "object",
+      "id" : "urn:jsonschema:org:apache:flink:runtime:messages:webmonitor:MultipleApplicationsDetails",
+      "properties" : {
+        "applications" : {
+          "type" : "array",
+          "items" : {
+            "type" : "object",
+            "id" : "urn:jsonschema:org:apache:flink:runtime:messages:webmonitor:ApplicationDetails",
+            "properties" : {
+              "id" : {
+                "type" : "any"
+              },
+              "name" : {
+                "type" : "string"
+              },
+              "start-time" : {
+                "type" : "integer"
+              },
+              "end-time" : {
+                "type" : "integer"
+              },
+              "duration" : {
+                "type" : "integer"
+              },
+              "status" : {
+                "type" : "string"
+              },
+              "jobs" : {
+                "type" : "object",
+                "additionalProperties" : {
+                  "type" : "integer"
+                }
+              }
+            }
+          }
+        }
+      }
+    }
+  }, {
+    "url" : "/applications/:applicationid",
+    "method" : "GET",
+    "status-code" : "200 OK",
+    "file-upload" : false,
+    "path-parameters" : {
+      "pathParameters" : [ {
+        "key" : "applicationid"
+      } ]
+    },
+    "query-parameters" : {
+      "queryParameters" : [ ]
+    },
+    "request" : {
+      "type" : "object",
+      "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:EmptyRequestBody"
+    },
+    "response" : {
+      "type" : "object",
+      "id" : "urn:jsonschema:org:apache:flink:runtime:messages:webmonitor:ApplicationDetailsInfo",
+      "properties" : {
+        "id" : {
+          "type" : "any"
+        },
+        "name" : {
+          "type" : "string"
+        },
+        "status" : {
+          "type" : "string"
+        },
+        "start-time" : {
+          "type" : "integer"
+        },
+        "end-time" : {
+          "type" : "integer"
+        },
+        "duration" : {
+          "type" : "integer"
+        },
+        "timestamps" : {
+          "type" : "object",
+          "additionalProperties" : {
+            "type" : "integer"
+          }
+        },
+        "jobs" : {
+          "type" : "array",
+          "items" : {
+            "type" : "object",
+            "id" : "urn:jsonschema:org:apache:flink:runtime:messages:webmonitor:JobDetails",
+            "properties" : {
+              "jid" : {
+                "type" : "any"
+              },
+              "name" : {
+                "type" : "string"
+              },
+              "start-time" : {
+                "type" : "integer"
+              },
+              "end-time" : {
+                "type" : "integer"
+              },
+              "duration" : {
+                "type" : "integer"
+              },
+              "state" : {
+                "type" : "string",
+                "enum" : [ "INITIALIZING", "CREATED", "RUNNING", "FAILING", "FAILED", "CANCELLING", "CANCELED", "FINISHED", "RESTARTING", "SUSPENDED", "RECONCILING" ]
+              },
+              "last-modification" : {
+                "type" : "integer"
+              },
+              "tasks" : {
+                "type" : "object",
+                "additionalProperties" : {
+                  "type" : "integer"
+                }
+              },
+              "pending-operators" : {
+                "type" : "integer"
+              }
+            }
+          }
+        }
+      }
+    }
+  }, {
+    "url" : "/applications/:applicationid/cancel",
+    "method" : "POST",
+    "status-code" : "202 Accepted",
+    "file-upload" : false,
+    "path-parameters" : {
+      "pathParameters" : [ {
+        "key" : "applicationid"
+      } ]
+    },
+    "query-parameters" : {
+      "queryParameters" : [ ]
+    },
+    "request" : {
+      "type" : "object",
+      "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:EmptyRequestBody"
+    },
+    "response" : {
+      "type" : "object",
+      "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:EmptyResponseBody"
+    }
+  }, {
     "url" : "/cluster",
     "method" : "DELETE",
     "status-code" : "200 OK",

flink-runtime/src/main/java/org/apache/flink/runtime/application/ArchivedApplication.java

@@ -0,0 +1,118 @@
+/*
+ * 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.flink.runtime.application;
+
+import org.apache.flink.api.common.ApplicationID;
+import org.apache.flink.api.common.ApplicationState;
+import org.apache.flink.runtime.executiongraph.ArchivedExecutionGraph;
+
+import java.io.Serializable;
+import java.util.Arrays;
+import java.util.Collection;
+
+/** Read-only information about an {@link AbstractApplication}. */
+public class ArchivedApplication implements Serializable {
+
+    private static final long serialVersionUID = 7231383912742578429L;
+
+    private final ApplicationID applicationId;
+
+    private final String applicationName;
+
+    private final ApplicationState applicationState;
+
+    private final long[] statusTimestamps;
+
+    private final Collection<ArchivedExecutionGraph> jobs;
+
+    public ArchivedApplication(
+            ApplicationID applicationId,
+            String applicationName,
+            ApplicationState applicationState,
+            long[] statusTimestamps,
+            Collection<ArchivedExecutionGraph> jobs) {
+        this.applicationId = applicationId;
+        this.applicationName = applicationName;
+        this.applicationState = applicationState;
+        this.statusTimestamps = statusTimestamps;
+        this.jobs = jobs;
+    }
+
+    public ApplicationID getApplicationId() {
+        return applicationId;
+    }
+
+    public String getApplicationName() {
+        return applicationName;
+    }
+
+    public ApplicationState getApplicationStatus() {
+        return applicationState;
+    }
+
+    public long getStatusTimestamp(ApplicationState status) {
+        return this.statusTimestamps[status.ordinal()];
+    }
+
+    public Collection<ArchivedExecutionGraph> getJobs() {
+        return jobs;
+    }
+
+    @Override
+    public String toString() {
+        return "ArchivedApplication{"
+                + "applicationId="
+                + applicationId
+                + ", applicationName='"
+                + applicationName
+                + '\''
+                + ", applicationState="
+                + applicationState
+                + ", statusTimestamps="
+                + Arrays.toString(statusTimestamps)
+                + ", jobs="
+                + jobs
+                + '}';
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        ArchivedApplication that = (ArchivedApplication) o;
+        return applicationId.equals(that.applicationId)
+                && applicationName.equals(that.applicationName)
+                && applicationState == that.applicationState
+                && Arrays.equals(statusTimestamps, that.statusTimestamps)
+                && jobs.equals(that.jobs);
+    }
+
+    @Override
+    public int hashCode() {
+        return 31 * applicationId.hashCode()
+                + 31 * applicationName.hashCode()
+                + 31 * applicationState.hashCode()
+                + 31 * Arrays.hashCode(statusTimestamps)
+                + 31 * jobs.hashCode();
+    }
+}