You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by uc...@apache.org on 2024/05/30 13:59:51 UTC
(flink) branch release-1.19 updated (663eeaf4092 -> 82dd642d620)
This is an automated email from the ASF dual-hosted git repository.
uce pushed a change to branch release-1.19
in repository https://gitbox.apache.org/repos/asf/flink.git
from 663eeaf4092 [FLINK-34379][table] Fix adding catalogtable logic
new f6caf4e6111 [FLINK-26808][rest] Add MultipartRoutes utility
new f567ee11cae [FLINK-26808][rest] Factor out getHandlerRoutes(RestHandlerSpecification)
new 82dd642d620 [FLINK-26808][rest] Only accept file upload at multipart routes
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:
.../flink/runtime/rest/FileUploadHandler.java | 31 ++++++-
.../flink/runtime/rest/RestServerEndpoint.java | 72 +++++++++-----
.../rest/handler/router/MultipartRoutes.java | 103 +++++++++++++++++++++
.../runtime/rest/handler/router/PathPattern.java | 5 +
.../flink/runtime/rest/handler/router/Router.java | 2 +-
.../runtime/rest/RestServerEndpointITCase.java | 77 ++++++++++-----
.../rest/handler/router/MultipartRoutesTest.java | 54 +++++++++++
7 files changed, 297 insertions(+), 47 deletions(-)
create mode 100644 flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/router/MultipartRoutes.java
create mode 100644 flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/router/MultipartRoutesTest.java
(flink) 01/03: [FLINK-26808][rest] Add MultipartRoutes utility
Posted by uc...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
uce pushed a commit to branch release-1.19
in repository https://gitbox.apache.org/repos/asf/flink.git
commit f6caf4e61114a520a20dff539111c23d36122fe7
Author: Ufuk Celebi <17...@users.noreply.github.com>
AuthorDate: Tue May 28 09:21:30 2024 +0200
[FLINK-26808][rest] Add MultipartRoutes utility
---
.../rest/handler/router/MultipartRoutes.java | 103 +++++++++++++++++++++
.../runtime/rest/handler/router/PathPattern.java | 5 +
.../flink/runtime/rest/handler/router/Router.java | 2 +-
.../rest/handler/router/MultipartRoutesTest.java | 54 +++++++++++
4 files changed, 163 insertions(+), 1 deletion(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/router/MultipartRoutes.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/router/MultipartRoutes.java
new file mode 100644
index 00000000000..46d4d327483
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/router/MultipartRoutes.java
@@ -0,0 +1,103 @@
+/*
+ * 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.router;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * A utility for {@link org.apache.flink.runtime.rest.FileUploadHandler} to determine whether it
+ * should accept a request.
+ */
+public class MultipartRoutes {
+
+ private final List<PathPattern> postRoutes;
+ private final List<PathPattern> fileUploadRoutes;
+
+ private MultipartRoutes(List<PathPattern> postRoutes, List<PathPattern> fileUploadRoutes) {
+ this.postRoutes = new ArrayList<>(postRoutes);
+ this.fileUploadRoutes = new ArrayList<>(fileUploadRoutes);
+ }
+
+ /**
+ * Returns <code>true</code> if the handler at the provided <code>requestUri</code> endpoint
+ * accepts POST requests.
+ *
+ * @param requestUri URI for the request
+ */
+ public boolean isPostRoute(String requestUri) {
+ return checkRoutes(requestUri, postRoutes);
+ }
+
+ /**
+ * Returns <code>true</code> if the handler at the provided <code>requestUri</code> endpoint
+ * accepts file uploads.
+ *
+ * @param requestUri URI for the request
+ */
+ public boolean isFileUploadRoute(String requestUri) {
+ return checkRoutes(requestUri, fileUploadRoutes);
+ }
+
+ @Override
+ public String toString() {
+ return "MultipartRoutes{"
+ + "postRoutes="
+ + postRoutes
+ + ", fileUploadRoutes="
+ + fileUploadRoutes
+ + '}';
+ }
+
+ private boolean checkRoutes(String requestUri, List<PathPattern> routes) {
+ String[] pathTokens = Router.decodePathTokens(requestUri);
+ Map<String, String> params = new HashMap<>();
+ for (PathPattern route : routes) {
+ if (route.match(pathTokens, params)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /** Builder for {@link MultipartRoutes}. */
+ public static class Builder {
+
+ private final List<PathPattern> postRoutes = new ArrayList<>();
+ private final List<PathPattern> fileUploadRoutes = new ArrayList<>();
+
+ public Builder() {}
+
+ public MultipartRoutes.Builder addPostRoute(String pattern) {
+ postRoutes.add(new PathPattern(pattern));
+ return this;
+ }
+
+ public MultipartRoutes.Builder addFileUploadRoute(String pattern) {
+ fileUploadRoutes.add(new PathPattern(pattern));
+ return this;
+ }
+
+ public MultipartRoutes build() {
+ return new MultipartRoutes(postRoutes, fileUploadRoutes);
+ }
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/router/PathPattern.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/router/PathPattern.java
index a2a3817922d..a1a0c91517d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/router/PathPattern.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/router/PathPattern.java
@@ -93,6 +93,11 @@ final class PathPattern {
return tokens;
}
+ @Override
+ public String toString() {
+ return pattern;
+ }
+
// --------------------------------------------------------------------------
// Instances of this class can be conveniently used as Map keys.
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/router/Router.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/router/Router.java
index 2faa37b9725..664b7d90ba9 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/router/Router.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/router/Router.java
@@ -235,7 +235,7 @@ public class Router<T> {
// --------------------------------------------------------------------------
- private String[] decodePathTokens(String uri) {
+ static String[] decodePathTokens(String uri) {
// Need to split the original URI (instead of QueryStringDecoder#path) then decode the
// tokens (components),
// otherwise /test1/123%2F456 will not match /test1/:p1
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/router/MultipartRoutesTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/router/MultipartRoutesTest.java
new file mode 100644
index 00000000000..b93cb981422
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/router/MultipartRoutesTest.java
@@ -0,0 +1,54 @@
+/*
+ * 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.router;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link MultipartRoutes}. */
+public class MultipartRoutesTest {
+
+ /** Test that multipart routes work with regular path patterns followed by {@link Router}. */
+ @Test
+ public void testRoutePatternMatching() {
+ MultipartRoutes.Builder builder = new MultipartRoutes.Builder();
+ builder.addPostRoute("/jobs");
+ builder.addPostRoute("/jobs/:jobid/stop");
+ builder.addPostRoute("/jobs/:jobid/savepoints/:savepointid/delete");
+
+ builder.addFileUploadRoute("/jobs");
+ builder.addFileUploadRoute("/jobs/:jobid/stop");
+ builder.addFileUploadRoute("/jobs/:jobid/savepoints/:savepointid/delete");
+
+ MultipartRoutes routes = builder.build();
+
+ assertThat(routes.isPostRoute("/jobs")).isTrue();
+ assertThat(routes.isPostRoute("/jobs?q1=p1&q2=p2")).isTrue();
+ assertThat(routes.isPostRoute("/jobs/abc")).isFalse();
+ assertThat(routes.isPostRoute("/jobs/abc/stop")).isTrue();
+ assertThat(routes.isPostRoute("/jobs/abc/savepoints/def/delete")).isTrue();
+
+ assertThat(routes.isFileUploadRoute("/jobs")).isTrue();
+ assertThat(routes.isFileUploadRoute("/jobs?q1=p1&q2=p2")).isTrue();
+ assertThat(routes.isFileUploadRoute("/jobs/abc")).isFalse();
+ assertThat(routes.isFileUploadRoute("/jobs/abc/stop")).isTrue();
+ assertThat(routes.isFileUploadRoute("/jobs/abc/savepoints/def/delete")).isTrue();
+ }
+}
(flink) 03/03: [FLINK-26808][rest] Only accept file upload at multipart routes
Posted by uc...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
uce pushed a commit to branch release-1.19
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 82dd642d6203da40af7939f7bf3c9cd4f9cc6fe1
Author: Ufuk Celebi <17...@users.noreply.github.com>
AuthorDate: Tue May 28 11:41:55 2024 +0200
[FLINK-26808][rest] Only accept file upload at multipart routes
---
.../flink/runtime/rest/FileUploadHandler.java | 31 ++++++++-
.../flink/runtime/rest/RestServerEndpoint.java | 33 +++++++++-
.../runtime/rest/RestServerEndpointITCase.java | 77 ++++++++++++++++------
3 files changed, 116 insertions(+), 25 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/FileUploadHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/FileUploadHandler.java
index c9e1fd78d74..3a969a57308 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/FileUploadHandler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/FileUploadHandler.java
@@ -19,6 +19,7 @@
package org.apache.flink.runtime.rest;
import org.apache.flink.runtime.rest.handler.FileUploads;
+import org.apache.flink.runtime.rest.handler.router.MultipartRoutes;
import org.apache.flink.runtime.rest.handler.util.HandlerUtils;
import org.apache.flink.runtime.rest.messages.ErrorResponseBody;
import org.apache.flink.runtime.rest.util.RestConstants;
@@ -79,15 +80,15 @@ public class FileUploadHandler extends SimpleChannelInboundHandler<HttpObject> {
private final Path uploadDir;
+ private final MultipartRoutes multipartRoutes;
+
private HttpPostRequestDecoder currentHttpPostRequestDecoder;
private HttpRequest currentHttpRequest;
private byte[] currentJsonPayload;
private Path currentUploadDir;
- private boolean addCRPrefix = false;
-
- public FileUploadHandler(final Path uploadDir) {
+ public FileUploadHandler(final Path uploadDir, final MultipartRoutes multipartRoutes) {
super(true);
// the clean up of temp files when jvm exits is handled by
@@ -103,6 +104,7 @@ public class FileUploadHandler extends SimpleChannelInboundHandler<HttpObject> {
DiskAttribute.baseDirectory = DiskFileUpload.baseDirectory;
this.uploadDir = requireNonNull(uploadDir);
+ this.multipartRoutes = requireNonNull(multipartRoutes);
}
@Override
@@ -125,6 +127,18 @@ public class FileUploadHandler extends SimpleChannelInboundHandler<HttpObject> {
new HttpPostRequestDecoder(DATA_FACTORY, httpRequest);
currentHttpRequest = ReferenceCountUtil.retain(httpRequest);
+ // We check this after initializing the multipart file upload in order for
+ // handleError to work correctly.
+ if (!multipartRoutes.isPostRoute(httpRequest.uri())) {
+ LOG.trace("POST request not allowed for {}.", httpRequest.uri());
+ handleError(
+ ctx,
+ "POST request not allowed",
+ HttpResponseStatus.BAD_REQUEST,
+ null);
+ return;
+ }
+
// make sure that we still have a upload dir in case that it got deleted in
// the meanwhile
RestServerEndpoint.createUploadDir(uploadDir, LOG, false);
@@ -151,6 +165,17 @@ public class FileUploadHandler extends SimpleChannelInboundHandler<HttpObject> {
&& hasNext(currentHttpPostRequestDecoder)) {
final InterfaceHttpData data = currentHttpPostRequestDecoder.next();
if (data.getHttpDataType() == InterfaceHttpData.HttpDataType.FileUpload) {
+ HttpRequest httpRequest = currentHttpRequest;
+ if (!multipartRoutes.isFileUploadRoute(httpRequest.uri())) {
+ LOG.trace("File upload not allowed for {}.", httpRequest.uri());
+ handleError(
+ ctx,
+ "File upload not allowed",
+ HttpResponseStatus.BAD_REQUEST,
+ null);
+ return;
+ }
+
final DiskFileUpload fileUpload = (DiskFileUpload) data;
checkState(fileUpload.isCompleted());
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestServerEndpoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestServerEndpoint.java
index 817e25521d8..4d021e019d0 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestServerEndpoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestServerEndpoint.java
@@ -28,8 +28,10 @@ import org.apache.flink.runtime.io.network.netty.SSLHandlerFactory;
import org.apache.flink.runtime.net.RedirectingSslHandler;
import org.apache.flink.runtime.rest.handler.PipelineErrorHandler;
import org.apache.flink.runtime.rest.handler.RestHandlerSpecification;
+import org.apache.flink.runtime.rest.handler.router.MultipartRoutes;
import org.apache.flink.runtime.rest.handler.router.Router;
import org.apache.flink.runtime.rest.handler.router.RouterHandler;
+import org.apache.flink.runtime.rest.messages.UntypedResponseMessageHeaders;
import org.apache.flink.runtime.rest.versioning.RestAPIVersion;
import org.apache.flink.util.AutoCloseableAsync;
import org.apache.flink.util.ConfigurationException;
@@ -196,6 +198,9 @@ public abstract class RestServerEndpoint implements RestService {
checkAllEndpointsAndHandlersAreUnique(handlers);
handlers.forEach(handler -> registerHandler(router, handler, log));
+ MultipartRoutes multipartRoutes = createMultipartRoutes(handlers);
+ log.debug("Using {} for FileUploadHandler", multipartRoutes);
+
ChannelInitializer<SocketChannel> initializer =
new ChannelInitializer<SocketChannel>() {
@@ -216,7 +221,7 @@ public abstract class RestServerEndpoint implements RestService {
ch.pipeline()
.addLast(new HttpServerCodec())
- .addLast(new FileUploadHandler(uploadDir))
+ .addLast(new FileUploadHandler(uploadDir, multipartRoutes))
.addLast(
new FlinkHttpObjectAggregator(
maxContentLength, responseHeaders));
@@ -635,6 +640,32 @@ public abstract class RestServerEndpoint implements RestService {
}
}
+ private MultipartRoutes createMultipartRoutes(
+ List<Tuple2<RestHandlerSpecification, ChannelInboundHandler>> handlers) {
+ MultipartRoutes.Builder builder = new MultipartRoutes.Builder();
+
+ for (Tuple2<RestHandlerSpecification, ChannelInboundHandler> handler : handlers) {
+ if (handler.f0.getHttpMethod() == HttpMethodWrapper.POST) {
+ for (String url : getHandlerRoutes(handler.f0)) {
+ builder.addPostRoute(url);
+ }
+ }
+
+ // The cast is necessary, because currently only UntypedResponseMessageHeaders exposes
+ // whether the handler accepts file uploads.
+ if (handler.f0 instanceof UntypedResponseMessageHeaders) {
+ UntypedResponseMessageHeaders<?, ?> header =
+ (UntypedResponseMessageHeaders<?, ?>) handler.f0;
+ if (header.acceptsFileUploads()) {
+ for (String url : getHandlerRoutes(header)) {
+ builder.addFileUploadRoute(url);
+ }
+ }
+ }
+ }
+ return builder.build();
+ }
+
private static Iterable<String> getHandlerRoutes(RestHandlerSpecification handlerSpec) {
final List<String> registeredRoutes = new ArrayList<>();
final String handlerUrl = handlerSpec.getTargetRestEndpointURL();
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 2c6d3b0afef..bcc1ee07f0e 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
@@ -374,25 +374,12 @@ public class RestServerEndpointITCase {
@TestTemplate
void testFileUpload() throws Exception {
final String boundary = generateMultiPartBoundary();
- final String crlf = "\r\n";
final String uploadedContent = "hello";
- final HttpURLConnection connection = openHttpConnectionForUpload(boundary);
-
- try (OutputStream output = connection.getOutputStream();
- PrintWriter writer =
- new PrintWriter(
- new OutputStreamWriter(output, StandardCharsets.UTF_8), true)) {
+ final HttpURLConnection connection =
+ openHttpConnectionForUpload(
+ boundary, TestUploadHeaders.INSTANCE.getTargetRestEndpointURL());
- writer.append("--" + boundary).append(crlf);
- writer.append("Content-Disposition: form-data; name=\"foo\"; filename=\"bar\"")
- .append(crlf);
- writer.append("Content-Type: plain/text; charset=utf8").append(crlf);
- writer.append(crlf).flush();
- output.write(uploadedContent.getBytes(StandardCharsets.UTF_8));
- output.flush();
- writer.append(crlf).flush();
- writer.append("--" + boundary + "--").append(crlf).flush();
- }
+ uploadFile(connection, uploadedContent, boundary);
assertThat(connection.getResponseCode()).isEqualTo(200);
final byte[] lastUploadedFileContents = testUploadHandler.getLastUploadedFileContents();
@@ -400,6 +387,32 @@ public class RestServerEndpointITCase {
.isEqualTo(new String(lastUploadedFileContents, StandardCharsets.UTF_8));
}
+ /**
+ * Tests that when a handler is marked as not accepting file uploads we (1) return an error and
+ * (2) don't upload the file to the upload directory.
+ */
+ @TestTemplate
+ void testFileUploadLimitedToAllowedUris() throws Exception {
+ final String boundary = generateMultiPartBoundary();
+ final File uploadDir = new File(tempFolder.toString(), "flink-web-upload");
+ final File[] preUploadFiles = uploadDir.listFiles();
+
+ // We need a handler that does not accept file uploads for this test
+ assertThat(TestVersionHeaders.INSTANCE.acceptsFileUploads()).isFalse();
+ String uri = TestVersionHeaders.INSTANCE.getTargetRestEndpointURL();
+
+ final HttpURLConnection connection = openHttpConnectionForUpload(boundary, uri);
+
+ uploadFile(connection, "hello", boundary);
+
+ assertThat(connection.getResponseCode()).isEqualTo(400);
+
+ // This is the important check. We don't want additional files when the handler does
+ // not accept file uploads.
+ final File[] postUploadFiles = uploadDir.listFiles();
+ assertThat(postUploadFiles).isEqualTo(preUploadFiles);
+ }
+
/**
* Sending multipart/form-data without a file should result in a bad request if the handler
* expects a file upload.
@@ -408,7 +421,9 @@ public class RestServerEndpointITCase {
void testMultiPartFormDataWithoutFileUpload() throws Exception {
final String boundary = generateMultiPartBoundary();
final String crlf = "\r\n";
- final HttpURLConnection connection = openHttpConnectionForUpload(boundary);
+ final HttpURLConnection connection =
+ openHttpConnectionForUpload(
+ boundary, TestUploadHeaders.INSTANCE.getTargetRestEndpointURL());
try (OutputStream output = connection.getOutputStream();
PrintWriter writer =
@@ -715,11 +730,11 @@ public class RestServerEndpointITCase {
return new File(resource.getFile());
}
- private HttpURLConnection openHttpConnectionForUpload(final String boundary)
- throws IOException {
+ private HttpURLConnection openHttpConnectionForUpload(
+ final String boundary, final String uploadUri) throws IOException {
final HttpURLConnection connection =
(HttpURLConnection)
- new URL(serverEndpoint.getRestBaseUrl() + "/upload").openConnection();
+ new URL(serverEndpoint.getRestBaseUrl() + uploadUri).openConnection();
connection.setDoOutput(true);
connection.setRequestProperty("Content-Type", "multipart/form-data; boundary=" + boundary);
return connection;
@@ -737,6 +752,26 @@ public class RestServerEndpointITCase {
return sb.toString();
}
+ private static void uploadFile(HttpURLConnection connection, String content, String boundary)
+ throws IOException {
+ final String crlf = "\r\n";
+ try (OutputStream output = connection.getOutputStream();
+ PrintWriter writer =
+ new PrintWriter(
+ new OutputStreamWriter(output, StandardCharsets.UTF_8), true)) {
+
+ writer.append("--" + boundary).append(crlf);
+ writer.append("Content-Disposition: form-data; name=\"foo\"; filename=\"bar\"")
+ .append(crlf);
+ writer.append("Content-Type: plain/text; charset=utf8").append(crlf);
+ writer.append(crlf).flush();
+ output.write(content.getBytes(StandardCharsets.UTF_8));
+ output.flush();
+ writer.append(crlf).flush();
+ writer.append("--" + boundary + "--").append(crlf).flush();
+ }
+ }
+
private static class TestHandler
extends AbstractRestHandler<RestfulGateway, TestRequest, TestResponse, TestParameters> {
(flink) 02/03: [FLINK-26808][rest] Factor out getHandlerRoutes(RestHandlerSpecification)
Posted by uc...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
uce pushed a commit to branch release-1.19
in repository https://gitbox.apache.org/repos/asf/flink.git
commit f567ee11caea2604efb1c36309fc83b00986dd51
Author: Ufuk Celebi <17...@users.noreply.github.com>
AuthorDate: Tue May 28 09:29:48 2024 +0200
[FLINK-26808][rest] Factor out getHandlerRoutes(RestHandlerSpecification)
---
.../flink/runtime/rest/RestServerEndpoint.java | 39 ++++++++++------------
1 file changed, 18 insertions(+), 21 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestServerEndpoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestServerEndpoint.java
index 004cb8106ec..817e25521d8 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestServerEndpoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestServerEndpoint.java
@@ -513,35 +513,17 @@ public abstract class RestServerEndpoint implements RestService {
Router router,
Tuple2<RestHandlerSpecification, ChannelInboundHandler> specificationHandler,
Logger log) {
- final String handlerURL = specificationHandler.f0.getTargetRestEndpointURL();
- // setup versioned urls
- for (final RestAPIVersion supportedVersion :
- specificationHandler.f0.getSupportedAPIVersions()) {
- final String versionedHandlerURL =
- '/' + supportedVersion.getURLVersionPrefix() + handlerURL;
+ for (String route : getHandlerRoutes(specificationHandler.f0)) {
log.debug(
"Register handler {} under {}@{}.",
specificationHandler.f1,
specificationHandler.f0.getHttpMethod(),
- versionedHandlerURL);
+ route);
registerHandler(
router,
- versionedHandlerURL,
+ route,
specificationHandler.f0.getHttpMethod(),
specificationHandler.f1);
- if (supportedVersion.isDefaultVersion()) {
- // setup unversioned url for convenience and backwards compatibility
- log.debug(
- "Register handler {} under {}@{}.",
- specificationHandler.f1,
- specificationHandler.f0.getHttpMethod(),
- handlerURL);
- registerHandler(
- router,
- handlerURL,
- specificationHandler.f0.getHttpMethod(),
- specificationHandler.f1);
- }
}
}
@@ -653,6 +635,21 @@ public abstract class RestServerEndpoint implements RestService {
}
}
+ private static Iterable<String> getHandlerRoutes(RestHandlerSpecification handlerSpec) {
+ final List<String> registeredRoutes = new ArrayList<>();
+ final String handlerUrl = handlerSpec.getTargetRestEndpointURL();
+ for (RestAPIVersion<?> supportedVersion : handlerSpec.getSupportedAPIVersions()) {
+ String versionedUrl = '/' + supportedVersion.getURLVersionPrefix() + handlerUrl;
+ registeredRoutes.add(versionedUrl);
+
+ if (supportedVersion.isDefaultVersion()) {
+ // Unversioned URL for convenience and backwards compatibility
+ registeredRoutes.add(handlerUrl);
+ }
+ }
+ return registeredRoutes;
+ }
+
/**
* Comparator for Rest URLs.
*