You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gu...@apache.org on 2022/03/17 02:02:07 UTC
[flink] branch master updated (d2d0065 -> 9851ad0)
This is an automated email from the ASF dual-hosted git repository.
guoyangze pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git.
from d2d0065 [FLINK-26607][python] Correct the MAX_LONG_VALUE/MIN_LONG_VALUE in several places
new 3a80185 [hotfix][doc] Re-generate rest docs
new ddc97b5 [FLINK-26641][rest] Introduce rest api to fetch job status
new 9851ad0 [FLINK-26641][client] Request the job status directly from the rest api
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../shortcodes/generated/rest_v1_dispatcher.html | 209 +++++++++++----------
docs/static/generated/rest_v1_dispatcher.yml | 58 ++++++
.../client/program/rest/RestClusterClient.java | 10 +-
.../client/program/rest/RestClusterClientTest.java | 49 ++---
.../src/test/resources/rest_api_v1.snapshot | 26 +++
.../runtime/messages/webmonitor/JobStatusInfo.java | 57 +++---
.../JobStatusHandler.java} | 24 +--
...tailsHeaders.java => JobStatusInfoHeaders.java} | 27 ++-
.../runtime/webmonitor/WebMonitorEndpoint.java | 10 +
.../webmonitor/JobStatusInfoTest.java} | 17 +-
.../rest/handler/job/JobStatusHandlerTest.java | 80 ++++++++
11 files changed, 371 insertions(+), 196 deletions(-)
copy flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/spec/PartitionSpec.java => flink-runtime/src/main/java/org/apache/flink/runtime/messages/webmonitor/JobStatusInfo.java (52%)
copy flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/{cluster/JobManagerThreadDumpHandler.java => job/JobStatusHandler.java} (69%)
copy flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/{JobDetailsHeaders.java => JobStatusInfoHeaders.java} (74%)
copy flink-runtime/src/test/java/org/apache/flink/runtime/{rest/handler/legacy/messages/ClusterOverviewWithVersionTest.java => messages/webmonitor/JobStatusInfoTest.java} (64%)
create mode 100644 flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/JobStatusHandlerTest.java
[flink] 02/03: [FLINK-26641][rest] Introduce rest api to fetch job status
Posted by gu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
guoyangze pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit ddc97b5772a89ab7a97b4b7d0572c93704216129
Author: Yangze Guo <ka...@gmail.com>
AuthorDate: Tue Mar 15 15:21:24 2022 +0800
[FLINK-26641][rest] Introduce rest api to fetch job status
---
.../shortcodes/generated/rest_v1_dispatcher.html | 71 +++++++++++++++++++
docs/static/generated/rest_v1_dispatcher.yml | 34 +++++++++
.../src/test/resources/rest_api_v1.snapshot | 26 +++++++
.../runtime/messages/webmonitor/JobStatusInfo.java | 70 +++++++++++++++++++
.../runtime/rest/handler/job/JobStatusHandler.java | 59 ++++++++++++++++
.../rest/messages/job/JobStatusInfoHeaders.java | 75 ++++++++++++++++++++
.../runtime/webmonitor/WebMonitorEndpoint.java | 10 +++
.../messages/webmonitor/JobStatusInfoTest.java | 35 ++++++++++
.../rest/handler/job/JobStatusHandlerTest.java | 80 ++++++++++++++++++++++
9 files changed, 460 insertions(+)
diff --git a/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html b/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html
index ca5386b..188f515 100644
--- a/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html
+++ b/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html
@@ -3507,6 +3507,77 @@ Using 'curl' you can upload a jar via 'curl -X POST -H "Expect:" -F "jarfile=@pa
<table class="rest-api table table-bordered">
<tbody>
<tr>
+ <td class="text-left" colspan="2"><h5><strong>/jobs/:jobid/status</strong></h5></td>
+ </tr>
+ <tr>
+ <td class="text-left" style="width: 20%">Verb: <code>GET</code></td>
+ <td class="text-left">Response code: <code>200 OK</code></td>
+ </tr>
+ <tr>
+ <td colspan="2">Returns the current status of a job execution.</td>
+ </tr>
+ <tr>
+ <td colspan="2">Path parameters</td>
+ </tr>
+ <tr>
+ <td colspan="2">
+ <ul>
+<li><code>jobid</code> - 32-character hexadecimal string value that identifies a job.</li>
+ </ul>
+ </td>
+ </tr>
+ <tr>
+ <td colspan="2">
+ <div class="book-expand">
+ <label>
+ <div class="book-expand-head flex justify-between">
+ <span>Request</span>
+ <span>▾</span>
+ </div>
+ <input type="checkbox" class="hidden">
+ <div class="book-expand-content markdown-inner">
+ <pre>
+ <code>
+{} </code>
+ </pre>
+ </div>
+ </label>
+ </div>
+ </td>
+ </tr>
+ <tr>
+ <td colspan="2">
+ <div class="book-expand">
+ <label>
+ <div class="book-expand-head flex justify-between">
+ <span>Response</span>
+ <span>▾</span>
+ </div>
+ <input type="checkbox" class="hidden">
+ <div class="book-expand-content markdown-inner">
+ <pre>
+ <code>
+{
+ "type" : "object",
+ "id" : "urn:jsonschema:org:apache:flink:runtime:messages:webmonitor:JobStatusInfo",
+ "properties" : {
+ "status" : {
+ "type" : "string",
+ "enum" : [ "INITIALIZING", "CREATED", "RUNNING", "FAILING", "FAILED", "CANCELLING", "CANCELED", "FINISHED", "RESTARTING", "SUSPENDED", "RECONCILING" ]
+ }
+ }
+} </code>
+ </pre>
+ </div>
+ </label>
+ </div>
+ </td>
+ </tr>
+ </tbody>
+</table>
+<table class="rest-api table table-bordered">
+ <tbody>
+ <tr>
<td class="text-left" colspan="2"><h5><strong>/jobs/:jobid/stop</strong></h5></td>
</tr>
<tr>
diff --git a/docs/static/generated/rest_v1_dispatcher.yml b/docs/static/generated/rest_v1_dispatcher.yml
index b3c3928..9b772f3 100644
--- a/docs/static/generated/rest_v1_dispatcher.yml
+++ b/docs/static/generated/rest_v1_dispatcher.yml
@@ -795,6 +795,23 @@ paths:
application/json:
schema:
$ref: '#/components/schemas/AsynchronousOperationResult'
+ /jobs/{jobid}/status:
+ get:
+ description: Returns the current status of a job execution.
+ parameters:
+ - name: jobid
+ in: path
+ description: 32-character hexadecimal string value that identifies a job.
+ required: true
+ schema:
+ $ref: '#/components/schemas/JobID'
+ responses:
+ "200":
+ description: The request was successful.
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/JobStatusInfo'
/jobs/{jobid}/stop:
post:
description: "Stops a job with a savepoint. Optionally, it can also emit a MAX_WATERMARK\
@@ -1841,6 +1858,23 @@ components:
num_acknowledged_subtasks:
type: integer
format: int32
+ JobStatusInfo:
+ type: object
+ properties:
+ status:
+ type: string
+ enum:
+ - INITIALIZING
+ - CREATED
+ - RUNNING
+ - FAILING
+ - FAILED
+ - CANCELLING
+ - CANCELED
+ - FINISHED
+ - RESTARTING
+ - SUSPENDED
+ - RECONCILING
JobAccumulator:
type: object
JobID:
diff --git a/flink-runtime-web/src/test/resources/rest_api_v1.snapshot b/flink-runtime-web/src/test/resources/rest_api_v1.snapshot
index 23a4a71..25c5b79 100644
--- a/flink-runtime-web/src/test/resources/rest_api_v1.snapshot
+++ b/flink-runtime-web/src/test/resources/rest_api_v1.snapshot
@@ -2003,6 +2003,32 @@
}
}
}, {
+ "url" : "/jobs/:jobid/status",
+ "method" : "GET",
+ "status-code" : "200 OK",
+ "file-upload" : false,
+ "path-parameters" : {
+ "pathParameters" : [ {
+ "key" : "jobid"
+ } ]
+ },
+ "query-parameters" : {
+ "queryParameters" : [ ]
+ },
+ "request" : {
+ "type" : "any"
+ },
+ "response" : {
+ "type" : "object",
+ "id" : "urn:jsonschema:org:apache:flink:runtime:messages:webmonitor:JobStatusInfo",
+ "properties" : {
+ "status" : {
+ "type" : "string",
+ "enum" : [ "INITIALIZING", "CREATED", "RUNNING", "FAILING", "FAILED", "CANCELLING", "CANCELED", "FINISHED", "RESTARTING", "SUSPENDED", "RECONCILING" ]
+ }
+ }
+ }
+ }, {
"url" : "/jobs/:jobid/stop",
"method" : "POST",
"status-code" : "202 Accepted",
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/messages/webmonitor/JobStatusInfo.java b/flink-runtime/src/main/java/org/apache/flink/runtime/messages/webmonitor/JobStatusInfo.java
new file mode 100644
index 0000000..9d1cf84
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/messages/webmonitor/JobStatusInfo.java
@@ -0,0 +1,70 @@
+/*
+ * 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.messages.webmonitor;
+
+import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.runtime.rest.messages.ResponseBody;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnore;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/** The status of a specific job. */
+public class JobStatusInfo implements ResponseBody, InfoMessage {
+ private static final long serialVersionUID = 1L;
+
+ public static final String FIELD_NAME_STATUS = "status";
+
+ @JsonProperty(FIELD_NAME_STATUS)
+ private final JobStatus jobStatus;
+
+ @JsonCreator
+ public JobStatusInfo(@JsonProperty(FIELD_NAME_STATUS) JobStatus jobStatus) {
+ this.jobStatus = checkNotNull(jobStatus);
+ }
+
+ @JsonIgnore
+ public JobStatus getJobStatus() {
+ return jobStatus;
+ }
+
+ @Override
+ public int hashCode() {
+ return jobStatus.hashCode();
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (obj == this) {
+ return true;
+ } else if (obj instanceof JobStatusInfo) {
+ JobStatusInfo that = (JobStatusInfo) obj;
+ return jobStatus.equals(that.jobStatus);
+ } else {
+ return false;
+ }
+ }
+
+ @Override
+ public String toString() {
+ return "JobStatusInfo { " + jobStatus + " }";
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobStatusHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobStatusHandler.java
new file mode 100644
index 0000000..4ca7ed7
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobStatusHandler.java
@@ -0,0 +1,59 @@
+/*
+ * 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.rest.handler.job;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.runtime.messages.webmonitor.JobStatusInfo;
+import org.apache.flink.runtime.rest.handler.AbstractRestHandler;
+import org.apache.flink.runtime.rest.handler.HandlerRequest;
+import org.apache.flink.runtime.rest.handler.RestHandlerException;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.JobIDPathParameter;
+import org.apache.flink.runtime.rest.messages.JobMessageParameters;
+import org.apache.flink.runtime.rest.messages.MessageHeaders;
+import org.apache.flink.runtime.webmonitor.RestfulGateway;
+import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
+
+import javax.annotation.Nonnull;
+
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+
+/** Handler for requesting job status. */
+public class JobStatusHandler
+ extends AbstractRestHandler<
+ RestfulGateway, EmptyRequestBody, JobStatusInfo, JobMessageParameters> {
+
+ public JobStatusHandler(
+ GatewayRetriever<? extends RestfulGateway> leaderRetriever,
+ Time timeout,
+ Map<String, String> responseHeaders,
+ MessageHeaders<EmptyRequestBody, JobStatusInfo, JobMessageParameters> messageHeaders) {
+ super(leaderRetriever, timeout, responseHeaders, messageHeaders);
+ }
+
+ @Override
+ protected CompletableFuture<JobStatusInfo> handleRequest(
+ @Nonnull HandlerRequest<EmptyRequestBody> request, @Nonnull RestfulGateway gateway)
+ throws RestHandlerException {
+ JobID jobId = request.getPathParameter(JobIDPathParameter.class);
+ return gateway.requestJobStatus(jobId, timeout).thenApply(JobStatusInfo::new);
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/JobStatusInfoHeaders.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/JobStatusInfoHeaders.java
new file mode 100644
index 0000000..22b76d7
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/JobStatusInfoHeaders.java
@@ -0,0 +1,75 @@
+/*
+ * 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.rest.messages.job;
+
+import org.apache.flink.runtime.messages.webmonitor.JobStatusInfo;
+import org.apache.flink.runtime.rest.HttpMethodWrapper;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.JobIDPathParameter;
+import org.apache.flink.runtime.rest.messages.JobMessageParameters;
+import org.apache.flink.runtime.rest.messages.MessageHeaders;
+
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+/**
+ * {@link MessageHeaders} for {@link org.apache.flink.runtime.rest.handler.job.JobStatusHandler}.
+ */
+public class JobStatusInfoHeaders
+ implements MessageHeaders<EmptyRequestBody, JobStatusInfo, JobMessageParameters> {
+ private static final JobStatusInfoHeaders INSTANCE = new JobStatusInfoHeaders();
+
+ @Override
+ public Class<EmptyRequestBody> getRequestClass() {
+ return EmptyRequestBody.class;
+ }
+
+ @Override
+ public Class<JobStatusInfo> getResponseClass() {
+ return JobStatusInfo.class;
+ }
+
+ @Override
+ public HttpResponseStatus getResponseStatusCode() {
+ return HttpResponseStatus.OK;
+ }
+
+ @Override
+ public JobMessageParameters getUnresolvedMessageParameters() {
+ return new JobMessageParameters();
+ }
+
+ @Override
+ public HttpMethodWrapper getHttpMethod() {
+ return HttpMethodWrapper.GET;
+ }
+
+ @Override
+ public String getTargetRestEndpointURL() {
+ return "/jobs/:" + JobIDPathParameter.KEY + "/status";
+ }
+
+ public static JobStatusInfoHeaders getInstance() {
+ return INSTANCE;
+ }
+
+ @Override
+ public String getDescription() {
+ return "Returns the current status of a job execution.";
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpoint.java
index 0178b11..3b21be9 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpoint.java
@@ -50,6 +50,7 @@ import org.apache.flink.runtime.rest.handler.job.JobExceptionsHandler;
import org.apache.flink.runtime.rest.handler.job.JobExecutionResultHandler;
import org.apache.flink.runtime.rest.handler.job.JobIdsHandler;
import org.apache.flink.runtime.rest.handler.job.JobPlanHandler;
+import org.apache.flink.runtime.rest.handler.job.JobStatusHandler;
import org.apache.flink.runtime.rest.handler.job.JobVertexAccumulatorsHandler;
import org.apache.flink.runtime.rest.handler.job.JobVertexBackPressureHandler;
import org.apache.flink.runtime.rest.handler.job.JobVertexDetailsHandler;
@@ -120,6 +121,7 @@ import org.apache.flink.runtime.rest.messages.cluster.JobManagerStdoutFileHeader
import org.apache.flink.runtime.rest.messages.cluster.JobManagerThreadDumpHeaders;
import org.apache.flink.runtime.rest.messages.cluster.ShutdownHeaders;
import org.apache.flink.runtime.rest.messages.job.JobDetailsHeaders;
+import org.apache.flink.runtime.rest.messages.job.JobStatusInfoHeaders;
import org.apache.flink.runtime.rest.messages.job.SubtaskCurrentAttemptDetailsHeaders;
import org.apache.flink.runtime.rest.messages.job.SubtaskExecutionAttemptAccumulatorsHeaders;
import org.apache.flink.runtime.rest.messages.job.SubtaskExecutionAttemptDetailsHeaders;
@@ -293,6 +295,13 @@ public class WebMonitorEndpoint<T extends RestfulGateway> extends RestServerEndp
responseHeaders,
JobIdsWithStatusesOverviewHeaders.getInstance());
+ JobStatusHandler jobStatusHandler =
+ new JobStatusHandler(
+ leaderRetriever,
+ timeout,
+ responseHeaders,
+ JobStatusInfoHeaders.getInstance());
+
JobsOverviewHandler jobsOverviewHandler =
new JobsOverviewHandler(
leaderRetriever,
@@ -653,6 +662,7 @@ public class WebMonitorEndpoint<T extends RestfulGateway> extends RestServerEndp
clusterConfigurationHandler));
handlers.add(Tuple2.of(dashboardConfigHandler.getMessageHeaders(), dashboardConfigHandler));
handlers.add(Tuple2.of(jobIdsHandler.getMessageHeaders(), jobIdsHandler));
+ handlers.add(Tuple2.of(jobStatusHandler.getMessageHeaders(), jobStatusHandler));
handlers.add(Tuple2.of(jobsOverviewHandler.getMessageHeaders(), jobsOverviewHandler));
handlers.add(Tuple2.of(jobConfigHandler.getMessageHeaders(), jobConfigHandler));
handlers.add(
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/messages/webmonitor/JobStatusInfoTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/messages/webmonitor/JobStatusInfoTest.java
new file mode 100644
index 0000000..93afec0
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/messages/webmonitor/JobStatusInfoTest.java
@@ -0,0 +1,35 @@
+/*
+ * 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.messages.webmonitor;
+
+import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.runtime.rest.messages.RestResponseMarshallingTestBase;
+
+/** Tests for the {@link JobStatusInfo}. */
+public class JobStatusInfoTest extends RestResponseMarshallingTestBase<JobStatusInfo> {
+ @Override
+ protected Class<JobStatusInfo> getTestResponseClass() {
+ return JobStatusInfo.class;
+ }
+
+ @Override
+ protected JobStatusInfo getTestResponseInstance() throws Exception {
+ return new JobStatusInfo(JobStatus.CREATED);
+ }
+}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/JobStatusHandlerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/JobStatusHandlerTest.java
new file mode 100644
index 0000000..a110cce
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/JobStatusHandlerTest.java
@@ -0,0 +1,80 @@
+/*
+ * 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.rest.handler.job;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.runtime.messages.webmonitor.JobStatusInfo;
+import org.apache.flink.runtime.rest.handler.HandlerRequest;
+import org.apache.flink.runtime.rest.handler.HandlerRequestException;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.JobIDPathParameter;
+import org.apache.flink.runtime.rest.messages.JobMessageParameters;
+import org.apache.flink.runtime.rest.messages.job.JobStatusInfoHeaders;
+import org.apache.flink.runtime.webmonitor.TestingRestfulGateway;
+import org.apache.flink.testutils.TestingUtils;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+
+import static org.junit.Assert.assertEquals;
+
+/** Test for the {@link JobStatusHandler}. */
+public class JobStatusHandlerTest extends TestLogger {
+ @Test
+ public void testRequestJobStatus() throws Exception {
+ final JobStatusHandler jobStatusHandler =
+ new JobStatusHandler(
+ CompletableFuture::new,
+ TestingUtils.TIMEOUT,
+ Collections.emptyMap(),
+ JobStatusInfoHeaders.getInstance());
+
+ final HandlerRequest<EmptyRequestBody> request = createRequest(new JobID());
+ final CompletableFuture<JobStatusInfo> response =
+ jobStatusHandler.handleRequest(
+ request,
+ new TestingRestfulGateway.Builder()
+ .setRequestJobStatusFunction(
+ ignored ->
+ CompletableFuture.completedFuture(
+ JobStatus.INITIALIZING))
+ .build());
+
+ assertEquals(response.get().getJobStatus(), JobStatus.INITIALIZING);
+ }
+
+ private static HandlerRequest<EmptyRequestBody> createRequest(JobID jobId)
+ throws HandlerRequestException {
+ final Map<String, String> pathParameters = new HashMap<>();
+ pathParameters.put(JobIDPathParameter.KEY, jobId.toString());
+
+ return HandlerRequest.resolveParametersAndCreate(
+ EmptyRequestBody.getInstance(),
+ new JobMessageParameters(),
+ pathParameters,
+ Collections.emptyMap(),
+ Collections.emptyList());
+ }
+}
[flink] 03/03: [FLINK-26641][client] Request the job status directly from the rest api
Posted by gu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
guoyangze pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 9851ad03c1b9eb529625b8f52c5c936b5b48cab7
Author: Yangze Guo <ka...@gmail.com>
AuthorDate: Tue Mar 15 16:50:02 2022 +0800
[FLINK-26641][client] Request the job status directly from the rest api
Before this commit, the client will fetch the current job status from JobDetails,
which might be stale for a few seconds because of the cache mechanism. As the job
status is lightweight enough, we fetch it directly from a specific job status fetcher
REST API which has no information delay. In this way, we shorten the time client waits
for the Job to finish initializing.
This closes #19095.
---
.../client/program/rest/RestClusterClient.java | 10 ++++-
.../client/program/rest/RestClusterClientTest.java | 49 +++++++---------------
2 files changed, 24 insertions(+), 35 deletions(-)
diff --git a/flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java b/flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java
index 656ffcf..ef8cc9b 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java
@@ -42,6 +42,7 @@ import org.apache.flink.runtime.jobgraph.OperatorID;
import org.apache.flink.runtime.jobmaster.JobResult;
import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService;
import org.apache.flink.runtime.messages.Acknowledge;
+import org.apache.flink.runtime.messages.webmonitor.JobStatusInfo;
import org.apache.flink.runtime.operators.coordination.CoordinationRequest;
import org.apache.flink.runtime.operators.coordination.CoordinationResponse;
import org.apache.flink.runtime.rest.FileUpload;
@@ -68,6 +69,7 @@ import org.apache.flink.runtime.rest.messages.cluster.ShutdownHeaders;
import org.apache.flink.runtime.rest.messages.job.JobDetailsHeaders;
import org.apache.flink.runtime.rest.messages.job.JobDetailsInfo;
import org.apache.flink.runtime.rest.messages.job.JobExecutionResultHeaders;
+import org.apache.flink.runtime.rest.messages.job.JobStatusInfoHeaders;
import org.apache.flink.runtime.rest.messages.job.JobSubmitHeaders;
import org.apache.flink.runtime.rest.messages.job.JobSubmitRequestBody;
import org.apache.flink.runtime.rest.messages.job.JobSubmitResponseBody;
@@ -744,8 +746,12 @@ public class RestClusterClient<T> implements ClusterClient<T> {
// -------------------------------------------------------------------------
private CompletableFuture<JobStatus> requestJobStatus(JobID jobId) {
- return getJobDetails(jobId)
- .thenApply(JobDetailsInfo::getJobStatus)
+ final JobStatusInfoHeaders jobStatusInfoHeaders = JobStatusInfoHeaders.getInstance();
+ final JobMessageParameters params = new JobMessageParameters();
+ params.jobPathParameter.resolve(jobId);
+
+ return sendRequest(jobStatusInfoHeaders, params)
+ .thenApply(JobStatusInfo::getJobStatus)
.thenApply(
jobStatus -> {
if (jobStatus == JobStatus.SUSPENDED) {
diff --git a/flink-clients/src/test/java/org/apache/flink/client/program/rest/RestClusterClientTest.java b/flink-clients/src/test/java/org/apache/flink/client/program/rest/RestClusterClientTest.java
index 80c67cb..1a86259 100644
--- a/flink-clients/src/test/java/org/apache/flink/client/program/rest/RestClusterClientTest.java
+++ b/flink-clients/src/test/java/org/apache/flink/client/program/rest/RestClusterClientTest.java
@@ -42,6 +42,7 @@ import org.apache.flink.runtime.jobgraph.OperatorID;
import org.apache.flink.runtime.jobmaster.JobResult;
import org.apache.flink.runtime.messages.Acknowledge;
import org.apache.flink.runtime.messages.webmonitor.JobDetails;
+import org.apache.flink.runtime.messages.webmonitor.JobStatusInfo;
import org.apache.flink.runtime.messages.webmonitor.MultipleJobsDetails;
import org.apache.flink.runtime.operators.coordination.CoordinationRequest;
import org.apache.flink.runtime.operators.coordination.CoordinationResponse;
@@ -71,10 +72,9 @@ import org.apache.flink.runtime.rest.messages.RequestBody;
import org.apache.flink.runtime.rest.messages.ResponseBody;
import org.apache.flink.runtime.rest.messages.TriggerId;
import org.apache.flink.runtime.rest.messages.TriggerIdPathParameter;
-import org.apache.flink.runtime.rest.messages.job.JobDetailsHeaders;
-import org.apache.flink.runtime.rest.messages.job.JobDetailsInfo;
import org.apache.flink.runtime.rest.messages.job.JobExecutionResultHeaders;
import org.apache.flink.runtime.rest.messages.job.JobExecutionResultResponseBody;
+import org.apache.flink.runtime.rest.messages.job.JobStatusInfoHeaders;
import org.apache.flink.runtime.rest.messages.job.JobSubmitHeaders;
import org.apache.flink.runtime.rest.messages.job.JobSubmitRequestBody;
import org.apache.flink.runtime.rest.messages.job.JobSubmitResponseBody;
@@ -924,11 +924,11 @@ public class RestClusterClientTest extends TestLogger {
*/
@Test
public void testNotShowSuspendedJobStatus() throws Exception {
- final List<JobDetailsInfo> jobDetails = new ArrayList<>();
- jobDetails.add(buildJobDetail(JobStatus.SUSPENDED));
- jobDetails.add(buildJobDetail(JobStatus.RUNNING));
+ final List<JobStatusInfo> jobStatusInfo = new ArrayList<>();
+ jobStatusInfo.add(new JobStatusInfo(JobStatus.SUSPENDED));
+ jobStatusInfo.add(new JobStatusInfo(JobStatus.RUNNING));
final TestJobStatusHandler jobStatusHandler =
- new TestJobStatusHandler(jobDetails.iterator());
+ new TestJobStatusHandler(jobStatusInfo.iterator());
try (TestRestServerEndpoint restServerEndpoint =
createRestServerEndpoint(jobStatusHandler)) {
@@ -943,23 +943,6 @@ public class RestClusterClientTest extends TestLogger {
}
}
- private JobDetailsInfo buildJobDetail(JobStatus jobStatus) {
- return new JobDetailsInfo(
- jobId,
- "testJob",
- true,
- jobStatus,
- 1L,
- 2L,
- 1L,
- 8888L,
- 1984L,
- new HashMap<>(),
- new ArrayList<>(),
- new HashMap<>(),
- "{\"id\":\"1234\"}");
- }
-
private class TestClientCoordinationHandler
extends TestHandler<
ClientCoordinationRequestBody,
@@ -1095,25 +1078,25 @@ public class RestClusterClientTest extends TestLogger {
}
private class TestJobStatusHandler
- extends TestHandler<EmptyRequestBody, JobDetailsInfo, JobMessageParameters> {
+ extends TestHandler<EmptyRequestBody, JobStatusInfo, JobMessageParameters> {
- private final Iterator<JobDetailsInfo> jobDetailsInfo;
+ private final Iterator<JobStatusInfo> jobStatusInfo;
- private TestJobStatusHandler(@Nonnull Iterator<JobDetailsInfo> jobDetailsInfo) {
- super(JobDetailsHeaders.getInstance());
- checkState(jobDetailsInfo.hasNext(), "Job details are empty");
- this.jobDetailsInfo = checkNotNull(jobDetailsInfo);
+ private TestJobStatusHandler(@Nonnull Iterator<JobStatusInfo> jobStatusInfo) {
+ super(JobStatusInfoHeaders.getInstance());
+ checkState(jobStatusInfo.hasNext(), "Job status are empty");
+ this.jobStatusInfo = checkNotNull(jobStatusInfo);
}
@Override
- protected CompletableFuture<JobDetailsInfo> handleRequest(
+ protected CompletableFuture<JobStatusInfo> handleRequest(
@Nonnull HandlerRequest<EmptyRequestBody> request,
@Nonnull DispatcherGateway gateway)
throws RestHandlerException {
- if (!jobDetailsInfo.hasNext()) {
- throw new IllegalStateException("More job details were requested than configured");
+ if (!jobStatusInfo.hasNext()) {
+ throw new IllegalStateException("More job status were requested than configured");
}
- return CompletableFuture.completedFuture(jobDetailsInfo.next());
+ return CompletableFuture.completedFuture(jobStatusInfo.next());
}
}
[flink] 01/03: [hotfix][doc] Re-generate rest docs
Posted by gu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
guoyangze pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 3a80185ed5f063022523bd83e58c1b6e1e040af4
Author: Yangze Guo <ka...@gmail.com>
AuthorDate: Tue Mar 15 16:27:40 2022 +0800
[hotfix][doc] Re-generate rest docs
---
.../shortcodes/generated/rest_v1_dispatcher.html | 138 +++++++--------------
docs/static/generated/rest_v1_dispatcher.yml | 24 ++++
2 files changed, 67 insertions(+), 95 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html b/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html
index e7c3784..ca5386b 100644
--- a/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html
+++ b/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html
@@ -1931,6 +1931,9 @@ Using 'curl' you can upload a jar via 'curl -X POST -H "Expect:" -F "jarfile=@pa
"type" : "string",
"enum" : [ "CHECKPOINT", "SAVEPOINT", "SYNC_SAVEPOINT" ]
},
+ "checkpointed_size" : {
+ "type" : "integer"
+ },
"end_to_end_duration" : {
"type" : "integer"
},
@@ -1955,9 +1958,6 @@ Using 'curl' you can upload a jar via 'curl -X POST -H "Expect:" -F "jarfile=@pa
"processed_data" : {
"type" : "integer"
},
- "checkpointed_size" : {
- "type" : "integer"
- },
"state_size" : {
"type" : "integer"
},
@@ -1993,6 +1993,9 @@ Using 'curl' you can upload a jar via 'curl -X POST -H "Expect:" -F "jarfile=@pa
"type" : "string",
"enum" : [ "CHECKPOINT", "SAVEPOINT", "SYNC_SAVEPOINT" ]
},
+ "checkpointed_size" : {
+ "type" : "integer"
+ },
"discarded" : {
"type" : "boolean"
},
@@ -2023,9 +2026,6 @@ Using 'curl' you can upload a jar via 'curl -X POST -H "Expect:" -F "jarfile=@pa
"processed_data" : {
"type" : "integer"
},
- "checkpointed_size" : {
- "type" : "integer"
- },
"state_size" : {
"type" : "integer"
},
@@ -2042,6 +2042,9 @@ Using 'curl' you can upload a jar via 'curl -X POST -H "Expect:" -F "jarfile=@pa
"alignment_buffered" : {
"type" : "integer"
},
+ "checkpointed_size" : {
+ "type" : "integer"
+ },
"end_to_end_duration" : {
"type" : "integer"
},
@@ -2063,9 +2066,6 @@ Using 'curl' you can upload a jar via 'curl -X POST -H "Expect:" -F "jarfile=@pa
"processed_data" : {
"type" : "integer"
},
- "checkpointed_size" : {
- "type" : "integer"
- },
"state_size" : {
"type" : "integer"
},
@@ -2092,6 +2092,9 @@ Using 'curl' you can upload a jar via 'curl -X POST -H "Expect:" -F "jarfile=@pa
"type" : "string",
"enum" : [ "CHECKPOINT", "SAVEPOINT", "SYNC_SAVEPOINT" ]
},
+ "checkpointed_size" : {
+ "type" : "integer"
+ },
"end_to_end_duration" : {
"type" : "integer"
},
@@ -2122,9 +2125,6 @@ Using 'curl' you can upload a jar via 'curl -X POST -H "Expect:" -F "jarfile=@pa
"processed_data" : {
"type" : "integer"
},
- "checkpointed_size" : {
- "type" : "integer"
- },
"state_size" : {
"type" : "integer"
},
@@ -2176,18 +2176,6 @@ Using 'curl' you can upload a jar via 'curl -X POST -H "Expect:" -F "jarfile=@pa
"type" : "object",
"$ref" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:checkpoints:StatsSummaryDto"
},
- "end_to_end_duration" : {
- "type" : "object",
- "$ref" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:checkpoints:StatsSummaryDto"
- },
- "persisted_data" : {
- "type" : "object",
- "$ref" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:checkpoints:StatsSummaryDto"
- },
- "processed_data" : {
- "type" : "object",
- "$ref" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:checkpoints:StatsSummaryDto"
- },
"checkpointed_size" : {
"type" : "object",
"id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:checkpoints:StatsSummaryDto",
@@ -2218,35 +2206,21 @@ Using 'curl' you can upload a jar via 'curl -X POST -H "Expect:" -F "jarfile=@pa
}
}
},
+ "end_to_end_duration" : {
+ "type" : "object",
+ "$ref" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:checkpoints:StatsSummaryDto"
+ },
+ "persisted_data" : {
+ "type" : "object",
+ "$ref" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:checkpoints:StatsSummaryDto"
+ },
+ "processed_data" : {
+ "type" : "object",
+ "$ref" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:checkpoints:StatsSummaryDto"
+ },
"state_size" : {
"type" : "object",
- "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:checkpoints:StatsSummaryDto",
- "properties" : {
- "avg" : {
- "type" : "integer"
- },
- "max" : {
- "type" : "integer"
- },
- "min" : {
- "type" : "integer"
- },
- "p50" : {
- "type" : "number"
- },
- "p90" : {
- "type" : "number"
- },
- "p95" : {
- "type" : "number"
- },
- "p99" : {
- "type" : "number"
- },
- "p999" : {
- "type" : "number"
- }
- }
+ "$ref" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:checkpoints:StatsSummaryDto"
}
}
}
@@ -2437,6 +2411,9 @@ Using 'curl' you can upload a jar via 'curl -X POST -H "Expect:" -F "jarfile=@pa
"type" : "string",
"enum" : [ "CHECKPOINT", "SAVEPOINT", "SYNC_SAVEPOINT" ]
},
+ "checkpointed_size" : {
+ "type" : "integer"
+ },
"end_to_end_duration" : {
"type" : "integer"
},
@@ -2461,9 +2438,6 @@ Using 'curl' you can upload a jar via 'curl -X POST -H "Expect:" -F "jarfile=@pa
"processed_data" : {
"type" : "integer"
},
- "checkpointed_size" : {
- "type" : "integer"
- },
"state_size" : {
"type" : "integer"
},
@@ -2480,6 +2454,9 @@ Using 'curl' you can upload a jar via 'curl -X POST -H "Expect:" -F "jarfile=@pa
"alignment_buffered" : {
"type" : "integer"
},
+ "checkpointed_size" : {
+ "type" : "integer"
+ },
"end_to_end_duration" : {
"type" : "integer"
},
@@ -2501,9 +2478,6 @@ Using 'curl' you can upload a jar via 'curl -X POST -H "Expect:" -F "jarfile=@pa
"processed_data" : {
"type" : "integer"
},
- "checkpointed_size" : {
- "type" : "integer"
- },
"state_size" : {
"type" : "integer"
},
@@ -2589,6 +2563,9 @@ Using 'curl' you can upload a jar via 'curl -X POST -H "Expect:" -F "jarfile=@pa
"alignment_buffered" : {
"type" : "integer"
},
+ "checkpointed_size" : {
+ "type" : "integer"
+ },
"end_to_end_duration" : {
"type" : "integer"
},
@@ -2610,9 +2587,6 @@ Using 'curl' you can upload a jar via 'curl -X POST -H "Expect:" -F "jarfile=@pa
"processed_data" : {
"type" : "integer"
},
- "checkpointed_size" : {
- "type" : "integer"
- },
"state_size" : {
"type" : "integer"
},
@@ -2675,14 +2649,6 @@ Using 'curl' you can upload a jar via 'curl -X POST -H "Expect:" -F "jarfile=@pa
}
}
},
- "end_to_end_duration" : {
- "type" : "object",
- "$ref" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:checkpoints:StatsSummaryDto"
- },
- "start_delay" : {
- "type" : "object",
- "$ref" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:checkpoints:StatsSummaryDto"
- },
"checkpointed_size" : {
"type" : "object",
"id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:checkpoints:StatsSummaryDto",
@@ -2713,35 +2679,17 @@ Using 'curl' you can upload a jar via 'curl -X POST -H "Expect:" -F "jarfile=@pa
}
}
},
+ "end_to_end_duration" : {
+ "type" : "object",
+ "$ref" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:checkpoints:StatsSummaryDto"
+ },
+ "start_delay" : {
+ "type" : "object",
+ "$ref" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:checkpoints:StatsSummaryDto"
+ },
"state_size" : {
"type" : "object",
- "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:checkpoints:StatsSummaryDto",
- "properties" : {
- "avg" : {
- "type" : "integer"
- },
- "max" : {
- "type" : "integer"
- },
- "min" : {
- "type" : "integer"
- },
- "p50" : {
- "type" : "number"
- },
- "p90" : {
- "type" : "number"
- },
- "p95" : {
- "type" : "number"
- },
- "p99" : {
- "type" : "number"
- },
- "p999" : {
- "type" : "number"
- }
- }
+ "$ref" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:checkpoints:StatsSummaryDto"
}
}
}
diff --git a/docs/static/generated/rest_v1_dispatcher.yml b/docs/static/generated/rest_v1_dispatcher.yml
index 80f7bee..b3c3928 100644
--- a/docs/static/generated/rest_v1_dispatcher.yml
+++ b/docs/static/generated/rest_v1_dispatcher.yml
@@ -1751,6 +1751,11 @@ components:
type: string
cancel-job:
type: boolean
+ formatType:
+ type: string
+ enum:
+ - CANONICAL
+ - NATIVE
triggerId:
$ref: '#/components/schemas/TriggerId'
ThreadDumpInfo:
@@ -1812,6 +1817,9 @@ components:
latest_ack_timestamp:
type: integer
format: int64
+ checkpointed_size:
+ type: integer
+ format: int64
state_size:
type: integer
format: int64
@@ -1860,6 +1868,9 @@ components:
latest_ack_timestamp:
type: integer
format: int64
+ checkpointed_size:
+ type: integer
+ format: int64
state_size:
type: integer
format: int64
@@ -2157,6 +2168,9 @@ components:
end_to_end_duration:
type: integer
format: int64
+ checkpointed_size:
+ type: integer
+ format: int64
state_size:
type: integer
format: int64
@@ -2368,6 +2382,11 @@ components:
type: string
drain:
type: boolean
+ formatType:
+ type: string
+ enum:
+ - CANONICAL
+ - NATIVE
triggerId:
$ref: '#/components/schemas/TriggerId'
DashboardConfiguration:
@@ -2604,6 +2623,9 @@ components:
latest_ack_timestamp:
type: integer
format: int64
+ checkpointed_size:
+ type: integer
+ format: int64
state_size:
type: integer
format: int64
@@ -2818,6 +2840,8 @@ components:
Summary:
type: object
properties:
+ checkpointed_size:
+ $ref: '#/components/schemas/StatsSummaryDto'
state_size:
$ref: '#/components/schemas/StatsSummaryDto'
end_to_end_duration: