This is an automated email from the ASF dual-hosted git repository. chesnay pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit d67338a140bf1b744d95a514b82824bba5b16105 Author: Chesnay Schepler <[email protected]> AuthorDate: Tue Jan 25 13:42:23 2022 +0100 [FLINK-25269][rest] Return 503 if rpc endpoint is not started --- .../flink/runtime/rpc/akka/AkkaRpcActor.java | 3 +- .../flink/runtime/rpc/akka/AkkaRpcActorTest.java | 6 +- .../exceptions/EndpointNotStartedException.java | 27 ++++++++ .../runtime/rest/handler/AbstractHandler.java | 9 +++ .../runtime/rest/RestServerEndpointITCase.java | 72 ++++++++++++++++++++++ 5 files changed, 113 insertions(+), 4 deletions(-) diff --git a/flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActor.java b/flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActor.java index 9f28e9a7f07..b4416e5cddf 100644 --- a/flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActor.java +++ b/flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActor.java @@ -25,6 +25,7 @@ import org.apache.flink.runtime.rpc.akka.exceptions.AkkaHandshakeException; import org.apache.flink.runtime.rpc.akka.exceptions.AkkaRpcException; import org.apache.flink.runtime.rpc.akka.exceptions.AkkaRpcInvalidStateException; import org.apache.flink.runtime.rpc.akka.exceptions.AkkaUnknownMessageException; +import org.apache.flink.runtime.rpc.exceptions.EndpointNotStartedException; import org.apache.flink.runtime.rpc.exceptions.RpcConnectionException; import org.apache.flink.runtime.rpc.messages.CallAsync; import org.apache.flink.runtime.rpc.messages.HandshakeSuccessMessage; @@ -171,7 +172,7 @@ class AkkaRpcActor<T extends RpcEndpoint & RpcGateway> extends AbstractActor { message); sendErrorIfSender( - new AkkaRpcException( + new EndpointNotStartedException( String.format( "Discard message %s, because the rpc endpoint %s has not been started yet.", message, rpcEndpoint.getAddress()))); diff --git a/flink-rpc/flink-rpc-akka/src/test/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActorTest.java b/flink-rpc/flink-rpc-akka/src/test/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActorTest.java index 80c62047e75..6f8c2bec8f6 100644 --- a/flink-rpc/flink-rpc-akka/src/test/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActorTest.java +++ b/flink-rpc/flink-rpc-akka/src/test/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActorTest.java @@ -30,7 +30,7 @@ import org.apache.flink.runtime.rpc.RpcService; import org.apache.flink.runtime.rpc.RpcServiceUtils; import org.apache.flink.runtime.rpc.RpcTimeout; import org.apache.flink.runtime.rpc.RpcUtils; -import org.apache.flink.runtime.rpc.akka.exceptions.AkkaRpcException; +import org.apache.flink.runtime.rpc.exceptions.EndpointNotStartedException; import org.apache.flink.runtime.rpc.exceptions.RecipientUnreachableException; import org.apache.flink.runtime.rpc.exceptions.RpcConnectionException; import org.apache.flink.runtime.rpc.exceptions.RpcException; @@ -135,9 +135,9 @@ class AkkaRpcActorTest { DummyRpcGateway rpcGateway = rpcEndpoint.getSelfGateway(DummyRpcGateway.class); - // this message should be discarded and completed with an AkkaRpcException + // this message should be discarded and complete with an exception assertThatThrownBy(() -> rpcGateway.foobar().get(timeout.getSize(), timeout.getUnit())) - .hasCauseInstanceOf(AkkaRpcException.class); + .hasCauseInstanceOf(EndpointNotStartedException.class); // set a new value which we expect to be returned rpcEndpoint.setFoobar(expectedValue); diff --git a/flink-rpc/flink-rpc-core/src/main/java/org/apache/flink/runtime/rpc/exceptions/EndpointNotStartedException.java b/flink-rpc/flink-rpc-core/src/main/java/org/apache/flink/runtime/rpc/exceptions/EndpointNotStartedException.java new file mode 100644 index 00000000000..d2b149a276b --- /dev/null +++ b/flink-rpc/flink-rpc-core/src/main/java/org/apache/flink/runtime/rpc/exceptions/EndpointNotStartedException.java @@ -0,0 +1,27 @@ +/* + * 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.rpc.exceptions; + +/** Exception which is thrown if an endpoint has not been started (yet). */ +public class EndpointNotStartedException extends RpcException { + private static final long serialVersionUID = -1725488032012148131L; + + public EndpointNotStartedException(String message) { + super(message); + } +} diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/AbstractHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/AbstractHandler.java index 897d732e750..b7d5c5bdca9 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/AbstractHandler.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/AbstractHandler.java @@ -29,6 +29,7 @@ import org.apache.flink.runtime.rest.messages.MessageParameters; import org.apache.flink.runtime.rest.messages.RequestBody; import org.apache.flink.runtime.rest.messages.UntypedResponseMessageHeaders; import org.apache.flink.runtime.rest.util.RestMapperUtils; +import org.apache.flink.runtime.rpc.exceptions.EndpointNotStartedException; import org.apache.flink.runtime.webmonitor.RestfulGateway; import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever; import org.apache.flink.util.AutoCloseableAsync; @@ -256,6 +257,14 @@ public abstract class AbstractHandler< new ErrorResponseBody(truncatedStackTrace), rhe.getHttpResponseStatus(), responseHeaders); + } else if (throwable instanceof EndpointNotStartedException) { + log.debug("A queried endpoint was not ready: {}", throwable.getMessage()); + return HandlerUtils.sendErrorResponse( + ctx, + httpRequest, + new ErrorResponseBody(throwable.getMessage()), + HttpResponseStatus.SERVICE_UNAVAILABLE, + responseHeaders); } else { log.error("Unhandled exception.", throwable); String stackTrace = diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/RestServerEndpointITCase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/RestServerEndpointITCase.java index 611b1978f6c..81c37a0efe1 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/RestServerEndpointITCase.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/RestServerEndpointITCase.java @@ -48,6 +48,7 @@ import org.apache.flink.runtime.rest.util.TestRestHandler; import org.apache.flink.runtime.rest.util.TestRestServerEndpoint; import org.apache.flink.runtime.rest.versioning.RestAPIVersion; import org.apache.flink.runtime.rpc.RpcUtils; +import org.apache.flink.runtime.rpc.exceptions.EndpointNotStartedException; import org.apache.flink.runtime.webmonitor.RestfulGateway; import org.apache.flink.runtime.webmonitor.TestingRestfulGateway; import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever; @@ -97,6 +98,7 @@ import java.nio.file.Path; import java.util.Arrays; import java.util.Collection; import java.util.Collections; +import java.util.Optional; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; import java.util.concurrent.ScheduledExecutorService; @@ -106,6 +108,7 @@ import java.util.stream.Collectors; import static java.util.Objects.requireNonNull; import static org.apache.flink.core.testutils.CommonTestUtils.assertThrows; +import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.hamcrest.CoreMatchers.hasItems; import static org.hamcrest.Matchers.containsString; import static org.hamcrest.Matchers.greaterThanOrEqualTo; @@ -246,6 +249,7 @@ public class RestServerEndpointITCase extends TestLogger { .withHandler( WebContentHandlerSpecification.getInstance(), staticFileServerHandler) + .withHandler(new TestUnavailableHandler(mockGatewayRetriever)) .buildAndStart(); restClient = new RestClient(config, EXECUTOR_RESOURCE.getExecutor()); @@ -703,6 +707,21 @@ public class RestServerEndpointITCase extends TestLogger { }); } + @Test + public void testOnUnavailableRpcEndpointReturns503() throws IOException { + CompletableFuture<EmptyResponseBody> response = + restClient.sendRequest( + serverAddress.getHostName(), + serverAddress.getPort(), + TestUnavailableHeaders.INSTANCE); + + assertThatThrownBy(response::get) + .extracting(x -> ExceptionUtils.findThrowable(x, RestClientException.class)) + .extracting(Optional::get) + .extracting(RestClientException::getHttpResponseStatus) + .isEqualTo(HttpResponseStatus.SERVICE_UNAVAILABLE); + } + private static File getTestResource(final String fileName) { final ClassLoader classLoader = ClassLoader.getSystemClassLoader(); final URL resource = classLoader.getResource(fileName); @@ -1163,4 +1182,57 @@ public class RestServerEndpointITCase extends TestLogger { return true; } } + + private enum TestUnavailableHeaders + implements MessageHeaders<EmptyRequestBody, EmptyResponseBody, EmptyMessageParameters> { + INSTANCE; + + @Override + public HttpMethodWrapper getHttpMethod() { + return HttpMethodWrapper.GET; + } + + @Override + public String getTargetRestEndpointURL() { + return "/unavailable"; + } + + @Override + public Class<EmptyRequestBody> getRequestClass() { + return EmptyRequestBody.class; + } + + @Override + public Class<EmptyResponseBody> getResponseClass() { + return EmptyResponseBody.class; + } + + @Override + public HttpResponseStatus getResponseStatusCode() { + return HttpResponseStatus.OK; + } + + @Override + public String getDescription() { + return ""; + } + + @Override + public EmptyMessageParameters getUnresolvedMessageParameters() { + return EmptyMessageParameters.getInstance(); + } + } + + private static class TestUnavailableHandler + extends TestRestHandler< + RestfulGateway, EmptyRequestBody, EmptyResponseBody, EmptyMessageParameters> { + + protected TestUnavailableHandler(GatewayRetriever<RestfulGateway> leaderRetriever) { + super( + leaderRetriever, + TestUnavailableHeaders.INSTANCE, + FutureUtils.completedExceptionally( + new EndpointNotStartedException("test exception"))); + } + } }
