[flink] 01/02: [FLINK-17634][rest] Reject multiple registration for the same endpoint

[ch...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

chesnay pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git

commit fa0362f430165f6ba9e25a50ce5d98e793b6ab38
Author: Chesnay Schepler <ch...@apache.org>
AuthorDate: Tue May 12 14:56:29 2020 +0200

    [FLINK-17634][rest] Reject multiple registration for the same endpoint
---
 .../flink/runtime/rest/RestServerEndpoint.java     | 43 +++++++++++++++++-----
 .../runtime/rest/RestServerEndpointITCase.java     | 21 ++++++++++-
 2 files changed, 54 insertions(+), 10 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 a264037..f575da2 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
@@ -64,6 +64,7 @@ import java.nio.file.Path;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.Comparator;
+import java.util.HashSet;
 import java.util.IdentityHashMap;
 import java.util.Iterator;
 import java.util.List;
@@ -140,7 +141,6 @@ public abstract class RestServerEndpoint implements AutoCloseableAsync {
 
 			final Router router = new Router();
 			final CompletableFuture<String> restAddressFuture = new CompletableFuture<>();
-			final Set<ChannelInboundHandler> distinctHandlers = Collections.newSetFromMap(new IdentityHashMap<>());
 
 			handlers = initializeHandlers(restAddressFuture);
 
@@ -155,14 +155,8 @@ public abstract class RestServerEndpoint implements AutoCloseableAsync {
 				handlers,
 				RestHandlerUrlComparator.INSTANCE);
 
-			handlers.forEach(handler -> {
-				if (distinctHandlers.contains(handler.f1)) {
-					throw new FlinkRuntimeException("Duplicate REST handler instance found."
-						+ " Please ensure each instance is registered only once.");
-				}
-				registerHandler(router, handler, log);
-				distinctHandlers.add(handler.f1);
-			});
+			checkAllEndpointsAndHandlersAreUnique(handlers);
+			handlers.forEach(handler -> registerHandler(router, handler, log));
 
 			ChannelInitializer<SocketChannel> initializer = new ChannelInitializer<SocketChannel>() {
 
@@ -491,6 +485,37 @@ public abstract class RestServerEndpoint implements AutoCloseableAsync {
 		}
 	}
 
+	private static void checkAllEndpointsAndHandlersAreUnique(final List<Tuple2<RestHandlerSpecification, ChannelInboundHandler>> handlers) {
+		// check for all handlers that
+		// 1) the instance is only registered once
+		// 2) only 1 handler is registered for each endpoint (defined by (version, method, url))
+		// technically the first check is redundant since a duplicate instance also returns the same headers which
+		// should fail the second check, but we get a better error message
+		final Set<String> uniqueEndpoints = new HashSet<>();
+		final Set<ChannelInboundHandler> distinctHandlers = Collections.newSetFromMap(new IdentityHashMap<>());
+		for (Tuple2<RestHandlerSpecification, ChannelInboundHandler> handler : handlers) {
+			boolean isNewHandler = distinctHandlers.add(handler.f1);
+			if (!isNewHandler) {
+				throw new FlinkRuntimeException("Duplicate REST handler instance found."
+					+ " Please ensure each instance is registered only once.");
+			}
+
+			final RestHandlerSpecification headers = handler.f0;
+			for (RestAPIVersion supportedAPIVersion : headers.getSupportedAPIVersions()) {
+				final String parameterizedEndpoint = supportedAPIVersion.toString() + headers.getHttpMethod() + headers.getTargetRestEndpointURL();
+				// normalize path parameters; distinct path parameters still clash at runtime
+				final String normalizedEndpoint = parameterizedEndpoint.replaceAll(":[\\w-]+", ":param");
+				boolean isNewEndpoint = uniqueEndpoints.add(normalizedEndpoint);
+				if (!isNewEndpoint) {
+					throw new FlinkRuntimeException(
+						String.format(
+							"REST handler registration overlaps with another registration for: version=%s, method=%s, url=%s.",
+							supportedAPIVersion, headers.getHttpMethod(), headers.getTargetRestEndpointURL()));
+				}
+			}
+		}
+	}
+
 	/**
 	 * Comparator for Rest URLs.
 	 *
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 0a95a3c..c941c45 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
@@ -595,12 +595,31 @@ public class RestServerEndpointITCase extends TestLogger {
 	}
 
 	@Test
+	public void testEndpointsMustBeUnique() throws Exception {
+		final RestServerEndpointConfiguration serverConfig = RestServerEndpointConfiguration.fromConfiguration(config);
+
+		final List<Tuple2<RestHandlerSpecification, ChannelInboundHandler>> handlers = Arrays.asList(
+			Tuple2.of(new TestHeaders(), testHandler),
+			Tuple2.of(new TestHeaders(), testUploadHandler)
+		);
+
+		assertThrows("REST handler registration",
+			FlinkRuntimeException.class,
+			() -> {
+				try (TestRestServerEndpoint restServerEndpoint =  new TestRestServerEndpoint(serverConfig, handlers)) {
+					restServerEndpoint.start();
+					return null;
+				}
+			});
+	}
+
+	@Test
 	public void testDuplicateHandlerRegistrationIsForbidden() throws Exception {
 		final RestServerEndpointConfiguration serverConfig = RestServerEndpointConfiguration.fromConfiguration(config);
 
 		final List<Tuple2<RestHandlerSpecification, ChannelInboundHandler>> handlers = Arrays.asList(
 			Tuple2.of(new TestHeaders(), testHandler),
-			Tuple2.of(new TestHeaders(), testHandler)
+			Tuple2.of(TestUploadHeaders.INSTANCE, testHandler)
 		);
 
 		assertThrows("Duplicate REST handler",