You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tr...@apache.org on 2018/02/23 09:24:37 UTC
[6/7] flink git commit: [FLINK-8635] [rest] Register rescaling
handlers at web endpoint
[FLINK-8635] [rest] Register rescaling handlers at web endpoint
This closes #5454.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/4e840979
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/4e840979
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/4e840979
Branch: refs/heads/master
Commit: 4e8409796a8fb80293db39d62745fbbcce6447cd
Parents: 06acdc1
Author: Till Rohrmann <tr...@apache.org>
Authored: Sun Feb 11 19:50:46 2018 +0100
Committer: Till Rohrmann <tr...@apache.org>
Committed: Thu Feb 22 17:32:38 2018 +0100
----------------------------------------------------------------------
...nchronousOperationTriggerMessageHeaders.java | 3 +-
.../rest/handler/job/RescalingHandlers.java | 111 ----------------
.../job/RescalingStatusMessageParameters.java | 39 ------
.../job/RescalingTriggerMessageParameters.java | 40 ------
.../job/rescaling/RescalingHandlers.java | 131 +++++++++++++++++++
.../job/rescaling/RescalingStatusHeaders.java | 78 +++++++++++
.../RescalingStatusMessageParameters.java | 40 ++++++
.../job/rescaling/RescalingTriggerHeaders.java | 70 ++++++++++
.../RescalingTriggerMessageParameters.java | 40 ++++++
.../RescaleParallelismQueryParameter.java | 41 ------
.../runtime/webmonitor/WebMonitorEndpoint.java | 22 +++-
11 files changed, 381 insertions(+), 234 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/4e840979/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/async/AsynchronousOperationTriggerMessageHeaders.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/async/AsynchronousOperationTriggerMessageHeaders.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/async/AsynchronousOperationTriggerMessageHeaders.java
index b317028..5baa5ad 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/async/AsynchronousOperationTriggerMessageHeaders.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/async/AsynchronousOperationTriggerMessageHeaders.java
@@ -28,7 +28,8 @@ import org.apache.flink.runtime.rest.messages.RequestBody;
* @param <R> type of the request
* @param <M> type of the message parameters
*/
-public abstract class AsynchronousOperationTriggerMessageHeaders<R extends RequestBody, M extends MessageParameters> implements MessageHeaders<R, TriggerResponse, M> {
+public abstract class AsynchronousOperationTriggerMessageHeaders<R extends RequestBody, M extends MessageParameters>
+ implements MessageHeaders<R, TriggerResponse, M> {
@Override
public Class<TriggerResponse> getResponseClass() {
http://git-wip-us.apache.org/repos/asf/flink/blob/4e840979/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/RescalingHandlers.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/RescalingHandlers.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/RescalingHandlers.java
deleted file mode 100644
index 6f8895a..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/RescalingHandlers.java
+++ /dev/null
@@ -1,111 +0,0 @@
-/*
- * 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.jobmaster.RescalingBehaviour;
-import org.apache.flink.runtime.messages.Acknowledge;
-import org.apache.flink.runtime.rest.handler.HandlerRequest;
-import org.apache.flink.runtime.rest.handler.RestHandlerException;
-import org.apache.flink.runtime.rest.handler.async.AbstractAsynchronousOperationHandlers;
-import org.apache.flink.runtime.rest.handler.async.AsynchronousOperationInfo;
-import org.apache.flink.runtime.rest.handler.async.AsynchronousOperationResult;
-import org.apache.flink.runtime.rest.handler.async.TriggerResponse;
-import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
-import org.apache.flink.runtime.rest.messages.JobIDPathParameter;
-import org.apache.flink.runtime.rest.messages.MessageHeaders;
-import org.apache.flink.runtime.rest.messages.RescaleParallelismQueryParameter;
-import org.apache.flink.runtime.rest.messages.TriggerId;
-import org.apache.flink.runtime.rest.messages.TriggerIdPathParameter;
-import org.apache.flink.runtime.webmonitor.RestfulGateway;
-import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
-import org.apache.flink.util.SerializedThrowable;
-
-import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
-
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.CompletableFuture;
-
-/**
- * Rest handler to trigger and poll the rescaling of a running job.
- */
-public class RescalingHandlers extends AbstractAsynchronousOperationHandlers<AsynchronousJobOperationKey, Acknowledge> {
-
- /**
- * Handler which triggers the rescaling of the specified job.
- */
- public class RescalingTriggerHandler extends TriggerHandler<RestfulGateway, EmptyRequestBody, RescalingTriggerMessageParameters> {
-
- protected RescalingTriggerHandler(CompletableFuture<String> localRestAddress, GatewayRetriever<? extends RestfulGateway> leaderRetriever, Time timeout, Map<String, String> responseHeaders, MessageHeaders<EmptyRequestBody, TriggerResponse, RescalingTriggerMessageParameters> messageHeaders) {
- super(localRestAddress, leaderRetriever, timeout, responseHeaders, messageHeaders);
- }
-
- @Override
- protected CompletableFuture<Acknowledge> triggerOperation(HandlerRequest<EmptyRequestBody, RescalingTriggerMessageParameters> request, RestfulGateway gateway) throws RestHandlerException {
- final JobID jobId = request.getPathParameter(JobIDPathParameter.class);
- final List<Integer> queryParameter = request.getQueryParameter(RescaleParallelismQueryParameter.class);
-
- if (queryParameter.isEmpty()) {
- throw new RestHandlerException("No new parallelism was specified.", HttpResponseStatus.BAD_REQUEST);
- }
-
- final int newParallelism = queryParameter.get(0);
-
- final CompletableFuture<Acknowledge> rescalingFuture = gateway.rescaleJob(jobId, newParallelism, RescalingBehaviour.STRICT, timeout);
-
- return rescalingFuture;
- }
-
- @Override
- protected AsynchronousJobOperationKey createOperationKey(HandlerRequest<EmptyRequestBody, RescalingTriggerMessageParameters> request) {
- final JobID jobId = request.getPathParameter(JobIDPathParameter.class);
- return AsynchronousJobOperationKey.of(new TriggerId(), jobId);
- }
- }
-
- /**
- * Handler which reports the status of the rescaling operation.
- */
- public class RescalingStatusHandler extends StatusHandler<RestfulGateway, AsynchronousOperationInfo, RescalingStatusMessageParameters> {
-
- protected RescalingStatusHandler(CompletableFuture<String> localRestAddress, GatewayRetriever<? extends RestfulGateway> leaderRetriever, Time timeout, Map<String, String> responseHeaders, MessageHeaders<EmptyRequestBody, AsynchronousOperationResult<AsynchronousOperationInfo>, RescalingStatusMessageParameters> messageHeaders) {
- super(localRestAddress, leaderRetriever, timeout, responseHeaders, messageHeaders);
- }
-
- @Override
- protected AsynchronousJobOperationKey getOperationKey(HandlerRequest<EmptyRequestBody, RescalingStatusMessageParameters> request) {
- final JobID jobId = request.getPathParameter(JobIDPathParameter.class);
- final TriggerId triggerId = request.getPathParameter(TriggerIdPathParameter.class);
-
- return AsynchronousJobOperationKey.of(triggerId, jobId);
- }
-
- @Override
- protected AsynchronousOperationInfo exceptionalOperationResultResponse(Throwable throwable) {
- return AsynchronousOperationInfo.completeExceptional(new SerializedThrowable(throwable));
- }
-
- @Override
- protected AsynchronousOperationInfo operationResultResponse(Acknowledge operationResult) {
- return AsynchronousOperationInfo.complete();
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/4e840979/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/RescalingStatusMessageParameters.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/RescalingStatusMessageParameters.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/RescalingStatusMessageParameters.java
deleted file mode 100644
index 4821b4f..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/RescalingStatusMessageParameters.java
+++ /dev/null
@@ -1,39 +0,0 @@
-/*
- * 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.runtime.rest.messages.JobMessageParameters;
-import org.apache.flink.runtime.rest.messages.MessagePathParameter;
-import org.apache.flink.runtime.rest.messages.TriggerIdPathParameter;
-
-import java.util.Arrays;
-import java.util.Collection;
-
-/**
- * Message headers for the {@link RescalingHandlers.RescalingStatusHandler}.
- */
-public class RescalingStatusMessageParameters extends JobMessageParameters {
-
- public final TriggerIdPathParameter triggerIdPathParameter = new TriggerIdPathParameter();
-
- @Override
- public Collection<MessagePathParameter<?>> getPathParameters() {
- return Arrays.asList(jobPathParameter, triggerIdPathParameter);
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/4e840979/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/RescalingTriggerMessageParameters.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/RescalingTriggerMessageParameters.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/RescalingTriggerMessageParameters.java
deleted file mode 100644
index 096baa1..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/RescalingTriggerMessageParameters.java
+++ /dev/null
@@ -1,40 +0,0 @@
-/*
- * 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.runtime.rest.messages.JobMessageParameters;
-import org.apache.flink.runtime.rest.messages.MessageParameters;
-import org.apache.flink.runtime.rest.messages.MessageQueryParameter;
-import org.apache.flink.runtime.rest.messages.RescalingParallelismQueryParameter;
-
-import java.util.Collection;
-import java.util.Collections;
-
-/**
- * {@link MessageParameters} for triggering the rescaling of a job.
- */
-public class RescalingTriggerMessageParameters extends JobMessageParameters {
-
- public final RescalingParallelismQueryParameter rescalingParallelismQueryParameter = new RescalingParallelismQueryParameter();
-
- @Override
- public Collection<MessageQueryParameter<?>> getQueryParameters() {
- return Collections.singleton(rescalingParallelismQueryParameter);
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/4e840979/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/rescaling/RescalingHandlers.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/rescaling/RescalingHandlers.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/rescaling/RescalingHandlers.java
new file mode 100644
index 0000000..3e4ae5a
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/rescaling/RescalingHandlers.java
@@ -0,0 +1,131 @@
+/*
+ * 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.rescaling;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.runtime.jobmaster.RescalingBehaviour;
+import org.apache.flink.runtime.messages.Acknowledge;
+import org.apache.flink.runtime.rest.handler.HandlerRequest;
+import org.apache.flink.runtime.rest.handler.RestHandlerException;
+import org.apache.flink.runtime.rest.handler.async.AbstractAsynchronousOperationHandlers;
+import org.apache.flink.runtime.rest.handler.async.AsynchronousOperationInfo;
+import org.apache.flink.runtime.rest.handler.job.AsynchronousJobOperationKey;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.JobIDPathParameter;
+import org.apache.flink.runtime.rest.messages.RescalingParallelismQueryParameter;
+import org.apache.flink.runtime.rest.messages.TriggerId;
+import org.apache.flink.runtime.rest.messages.TriggerIdPathParameter;
+import org.apache.flink.runtime.webmonitor.RestfulGateway;
+import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
+import org.apache.flink.util.SerializedThrowable;
+
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Rest handler to trigger and poll the rescaling of a running job.
+ */
+public class RescalingHandlers extends AbstractAsynchronousOperationHandlers<AsynchronousJobOperationKey, Acknowledge> {
+
+ /**
+ * Handler which triggers the rescaling of the specified job.
+ */
+ public class RescalingTriggerHandler extends TriggerHandler<RestfulGateway, EmptyRequestBody, RescalingTriggerMessageParameters> {
+
+ public RescalingTriggerHandler(
+ CompletableFuture<String> localRestAddress,
+ GatewayRetriever<? extends RestfulGateway> leaderRetriever,
+ Time timeout,
+ Map<String, String> responseHeaders) {
+ super(
+ localRestAddress,
+ leaderRetriever,
+ timeout,
+ responseHeaders,
+ RescalingTriggerHeaders.getInstance());
+ }
+
+ @Override
+ protected CompletableFuture<Acknowledge> triggerOperation(HandlerRequest<EmptyRequestBody, RescalingTriggerMessageParameters> request, RestfulGateway gateway) throws RestHandlerException {
+ final JobID jobId = request.getPathParameter(JobIDPathParameter.class);
+ final List<Integer> queryParameter = request.getQueryParameter(RescalingParallelismQueryParameter.class);
+
+ if (queryParameter.isEmpty()) {
+ throw new RestHandlerException("No new parallelism was specified.", HttpResponseStatus.BAD_REQUEST);
+ }
+
+ final int newParallelism = queryParameter.get(0);
+
+ final CompletableFuture<Acknowledge> rescalingFuture = gateway.rescaleJob(
+ jobId,
+ newParallelism,
+ RescalingBehaviour.STRICT,
+ timeout);
+
+ return rescalingFuture;
+ }
+
+ @Override
+ protected AsynchronousJobOperationKey createOperationKey(HandlerRequest<EmptyRequestBody, RescalingTriggerMessageParameters> request) {
+ final JobID jobId = request.getPathParameter(JobIDPathParameter.class);
+ return AsynchronousJobOperationKey.of(new TriggerId(), jobId);
+ }
+ }
+
+ /**
+ * Handler which reports the status of the rescaling operation.
+ */
+ public class RescalingStatusHandler extends StatusHandler<RestfulGateway, AsynchronousOperationInfo, RescalingStatusMessageParameters> {
+
+ public RescalingStatusHandler(
+ CompletableFuture<String> localRestAddress,
+ GatewayRetriever<? extends RestfulGateway> leaderRetriever,
+ Time timeout,
+ Map<String, String> responseHeaders) {
+ super(
+ localRestAddress,
+ leaderRetriever,
+ timeout,
+ responseHeaders,
+ RescalingStatusHeaders.getInstance());
+ }
+
+ @Override
+ protected AsynchronousJobOperationKey getOperationKey(HandlerRequest<EmptyRequestBody, RescalingStatusMessageParameters> request) {
+ final JobID jobId = request.getPathParameter(JobIDPathParameter.class);
+ final TriggerId triggerId = request.getPathParameter(TriggerIdPathParameter.class);
+
+ return AsynchronousJobOperationKey.of(triggerId, jobId);
+ }
+
+ @Override
+ protected AsynchronousOperationInfo exceptionalOperationResultResponse(Throwable throwable) {
+ return AsynchronousOperationInfo.completeExceptional(new SerializedThrowable(throwable));
+ }
+
+ @Override
+ protected AsynchronousOperationInfo operationResultResponse(Acknowledge operationResult) {
+ return AsynchronousOperationInfo.complete();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/4e840979/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/rescaling/RescalingStatusHeaders.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/rescaling/RescalingStatusHeaders.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/rescaling/RescalingStatusHeaders.java
new file mode 100644
index 0000000..2d5babb
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/rescaling/RescalingStatusHeaders.java
@@ -0,0 +1,78 @@
+/*
+ * 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.rescaling;
+
+import org.apache.flink.runtime.rest.HttpMethodWrapper;
+import org.apache.flink.runtime.rest.handler.async.AsynchronousOperationInfo;
+import org.apache.flink.runtime.rest.handler.async.AsynchronousOperationStatusMessageHeaders;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.JobIDPathParameter;
+import org.apache.flink.runtime.rest.messages.TriggerIdPathParameter;
+
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+/**
+ * Message headers for polling the status of an ongoing rescaling operation.
+ */
+public class RescalingStatusHeaders extends
+ AsynchronousOperationStatusMessageHeaders<AsynchronousOperationInfo, RescalingStatusMessageParameters> {
+
+ private static final RescalingStatusHeaders INSTANCE = new RescalingStatusHeaders();
+
+ private static final String URL = String.format(
+ "/jobs/:%s/rescaling/:%s",
+ JobIDPathParameter.KEY,
+ TriggerIdPathParameter.KEY);
+
+ private RescalingStatusHeaders() {}
+
+ @Override
+ protected Class<AsynchronousOperationInfo> getValueClass() {
+ return AsynchronousOperationInfo.class;
+ }
+
+ @Override
+ public Class<EmptyRequestBody> getRequestClass() {
+ return EmptyRequestBody.class;
+ }
+
+ @Override
+ public HttpResponseStatus getResponseStatusCode() {
+ return HttpResponseStatus.OK;
+ }
+
+ @Override
+ public RescalingStatusMessageParameters getUnresolvedMessageParameters() {
+ return new RescalingStatusMessageParameters();
+ }
+
+ @Override
+ public HttpMethodWrapper getHttpMethod() {
+ return HttpMethodWrapper.GET;
+ }
+
+ @Override
+ public String getTargetRestEndpointURL() {
+ return URL;
+ }
+
+ public static RescalingStatusHeaders getInstance() {
+ return INSTANCE;
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/4e840979/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/rescaling/RescalingStatusMessageParameters.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/rescaling/RescalingStatusMessageParameters.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/rescaling/RescalingStatusMessageParameters.java
new file mode 100644
index 0000000..938a363
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/rescaling/RescalingStatusMessageParameters.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.flink.runtime.rest.handler.job.rescaling;
+
+import org.apache.flink.runtime.rest.messages.JobMessageParameters;
+import org.apache.flink.runtime.rest.messages.MessageParameters;
+import org.apache.flink.runtime.rest.messages.MessagePathParameter;
+import org.apache.flink.runtime.rest.messages.TriggerIdPathParameter;
+
+import java.util.Arrays;
+import java.util.Collection;
+
+/**
+ * {@link MessageParameters} for polling the status of a rescaling operation.
+ */
+public class RescalingStatusMessageParameters extends JobMessageParameters {
+
+ public final TriggerIdPathParameter triggerIdPathParameter = new TriggerIdPathParameter();
+
+ @Override
+ public Collection<MessagePathParameter<?>> getPathParameters() {
+ return Arrays.asList(jobPathParameter, triggerIdPathParameter);
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/4e840979/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/rescaling/RescalingTriggerHeaders.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/rescaling/RescalingTriggerHeaders.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/rescaling/RescalingTriggerHeaders.java
new file mode 100644
index 0000000..9567410
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/rescaling/RescalingTriggerHeaders.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.rest.handler.job.rescaling;
+
+import org.apache.flink.runtime.rest.HttpMethodWrapper;
+import org.apache.flink.runtime.rest.handler.async.AsynchronousOperationTriggerMessageHeaders;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.JobIDPathParameter;
+
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+/**
+ * Message headers for triggering the rescaling of a job.
+ */
+public class RescalingTriggerHeaders extends
+ AsynchronousOperationTriggerMessageHeaders<EmptyRequestBody, RescalingTriggerMessageParameters> {
+
+ private static final RescalingTriggerHeaders INSTANCE = new RescalingTriggerHeaders();
+
+ private static final String URL = String.format(
+ "/jobs/:%s/rescaling",
+ JobIDPathParameter.KEY);
+
+ private RescalingTriggerHeaders() {}
+
+ @Override
+ public Class<EmptyRequestBody> getRequestClass() {
+ return EmptyRequestBody.class;
+ }
+
+ @Override
+ public HttpResponseStatus getResponseStatusCode() {
+ return HttpResponseStatus.OK;
+ }
+
+ @Override
+ public RescalingTriggerMessageParameters getUnresolvedMessageParameters() {
+ return new RescalingTriggerMessageParameters();
+ }
+
+ @Override
+ public HttpMethodWrapper getHttpMethod() {
+ return HttpMethodWrapper.PATCH;
+ }
+
+ @Override
+ public String getTargetRestEndpointURL() {
+ return URL;
+ }
+
+ public static RescalingTriggerHeaders getInstance() {
+ return INSTANCE;
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/4e840979/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/rescaling/RescalingTriggerMessageParameters.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/rescaling/RescalingTriggerMessageParameters.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/rescaling/RescalingTriggerMessageParameters.java
new file mode 100644
index 0000000..4b5d307
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/rescaling/RescalingTriggerMessageParameters.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.flink.runtime.rest.handler.job.rescaling;
+
+import org.apache.flink.runtime.rest.messages.JobMessageParameters;
+import org.apache.flink.runtime.rest.messages.MessageParameters;
+import org.apache.flink.runtime.rest.messages.MessageQueryParameter;
+import org.apache.flink.runtime.rest.messages.RescalingParallelismQueryParameter;
+
+import java.util.Collection;
+import java.util.Collections;
+
+/**
+ * {@link MessageParameters} for triggering the rescaling of a job.
+ */
+public class RescalingTriggerMessageParameters extends JobMessageParameters {
+
+ public final RescalingParallelismQueryParameter rescalingParallelismQueryParameter = new RescalingParallelismQueryParameter();
+
+ @Override
+ public Collection<MessageQueryParameter<?>> getQueryParameters() {
+ return Collections.singleton(rescalingParallelismQueryParameter);
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/4e840979/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/RescaleParallelismQueryParameter.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/RescaleParallelismQueryParameter.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/RescaleParallelismQueryParameter.java
deleted file mode 100644
index 8058ab9..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/RescaleParallelismQueryParameter.java
+++ /dev/null
@@ -1,41 +0,0 @@
-/*
- * 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;
-
-/**
- * Parallelism for the rescaling of jobs specified as a {@link MessageQueryParameter}.
- */
-public class RescaleParallelismQueryParameter extends MessageQueryParameter<Integer> {
-
- public static final String KEY = "parallelism";
-
- protected RescaleParallelismQueryParameter() {
- super(KEY, MessageParameterRequisiteness.MANDATORY);
- }
-
- @Override
- public Integer convertValueFromString(String value) {
- return Integer.valueOf(value);
- }
-
- @Override
- public String convertStringToValue(Integer value) {
- return value.toString();
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/4e840979/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpoint.java
----------------------------------------------------------------------
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 427332f..10a3650 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
@@ -61,6 +61,9 @@ import org.apache.flink.runtime.rest.handler.job.metrics.JobMetricsHandler;
import org.apache.flink.runtime.rest.handler.job.metrics.JobVertexMetricsHandler;
import org.apache.flink.runtime.rest.handler.job.metrics.SubtaskMetricsHandler;
import org.apache.flink.runtime.rest.handler.job.metrics.TaskManagerMetricsHandler;
+import org.apache.flink.runtime.rest.handler.job.rescaling.RescalingHandlers;
+import org.apache.flink.runtime.rest.handler.job.rescaling.RescalingStatusHeaders;
+import org.apache.flink.runtime.rest.handler.job.rescaling.RescalingTriggerHeaders;
import org.apache.flink.runtime.rest.handler.job.savepoints.SavepointHandlers;
import org.apache.flink.runtime.rest.handler.legacy.ConstantTextHandler;
import org.apache.flink.runtime.rest.handler.legacy.ExecutionGraphCache;
@@ -448,8 +451,21 @@ public class WebMonitorEndpoint<T extends RestfulGateway> extends RestServerEndp
SubtaskCurrentAttemptDetailsHeaders.getInstance(),
executionGraphCache,
executor,
- metricFetcher
- );
+ metricFetcher);
+
+ final RescalingHandlers rescalingHandlers = new RescalingHandlers();
+
+ final RescalingHandlers.RescalingTriggerHandler rescalingTriggerHandler = rescalingHandlers.new RescalingTriggerHandler(
+ restAddressFuture,
+ leaderRetriever,
+ timeout,
+ responseHeaders);
+
+ final RescalingHandlers.RescalingStatusHandler rescalingStatusHandler = rescalingHandlers.new RescalingStatusHandler(
+ restAddressFuture,
+ leaderRetriever,
+ timeout,
+ responseHeaders);
JobVertexBackPressureHandler jobVertexBackPressureHandler = new JobVertexBackPressureHandler(
restAddressFuture,
@@ -532,6 +548,8 @@ public class WebMonitorEndpoint<T extends RestfulGateway> extends RestServerEndp
handlers.add(Tuple2.of(JobVertexBackPressureHeaders.getInstance(), jobVertexBackPressureHandler));
handlers.add(Tuple2.of(JobTerminationHeaders.getInstance(), jobCancelTerminationHandler));
handlers.add(Tuple2.of(JobVertexDetailsHeaders.getInstance(), jobVertexDetailsHandler));
+ handlers.add(Tuple2.of(RescalingTriggerHeaders.getInstance(), rescalingTriggerHandler));
+ handlers.add(Tuple2.of(RescalingStatusHeaders.getInstance(), rescalingStatusHandler));
// TODO: Remove once the Yarn proxy can forward all REST verbs
handlers.add(Tuple2.of(YarnCancelJobTerminationHeaders.getInstance(), jobCancelTerminationHandler));