You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by pnowojski <gi...@git.apache.org> on 2018/05/17 10:24:51 UTC
[GitHub] flink pull request #6031: [FLINK-9386] Embed netty router
GitHub user pnowojski opened a pull request:
https://github.com/apache/flink/pull/6031
[FLINK-9386] Embed netty router
This replaces netty-router dependency with our own version of it, which is simplified and adds guarantees about order of matching router patterns.
This is a prerequisite for https://issues.apache.org/jira/browse/FLINK-3952 . netty-router 1.10 is incompatible with Netty 4.1, while netty-router 2.2.0 brakes a compatibility in a way that we were unable to use it.
## Verifying this change
This change ports part of the netty-router code, including tests and adds a test for preserving patter matching order by the `Router`. Additionally this change is covered by various of our ITCases.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (yes / **no**)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (yes / **no**)
- The serializers: (yes / **no** / don't know)
- The runtime per-record code paths (performance sensitive): (yes / **no** / don't know)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (yes / **no** / don't know)
- The S3 file system connector: (yes / **no** / don't know)
## Documentation
- Does this pull request introduce a new feature? (**yes** / no)
- If yes, how is the feature documented? (not applicable / docs / **JavaDocs** / not documented)
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/pnowojski/flink f9386
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/6031.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #6031
----
commit 923ef1db7d480c5263f385e87c1c6645fd54ce91
Author: Piotr Nowojski <pi...@...>
Date: 2018-05-14T11:46:08Z
[hotfix][tests] Report failure with error level instead of debug
commit 556fbbe283e1ea64783df73dbbfa3ddd91fc9ffe
Author: Piotr Nowojski <pi...@...>
Date: 2018-05-16T19:26:36Z
[FLINK-9386] Embed netty router
This commit replaces netty-router dependency with our own version of it, which is
simplified and adds guarantees about order of matching router patterns.
----
---
[GitHub] flink pull request #6031: [FLINK-9386] Embed netty router
Posted by uce <gi...@git.apache.org>.
Github user uce commented on a diff in the pull request:
https://github.com/apache/flink/pull/6031#discussion_r190274982
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/router/RouterHandler.java ---
@@ -0,0 +1,109 @@
+/*
+ * 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.apache.flink.runtime.rest.handler.AbstractRestHandler;
+import org.apache.flink.runtime.rest.handler.util.HandlerUtils;
+import org.apache.flink.runtime.rest.messages.ErrorResponseBody;
+
+import org.apache.flink.shaded.netty4.io.netty.channel.ChannelHandler;
+import org.apache.flink.shaded.netty4.io.netty.channel.ChannelHandlerContext;
+import org.apache.flink.shaded.netty4.io.netty.channel.ChannelInboundHandler;
+import org.apache.flink.shaded.netty4.io.netty.channel.ChannelPipeline;
+import org.apache.flink.shaded.netty4.io.netty.channel.SimpleChannelInboundHandler;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.DefaultFullHttpResponse;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpHeaders;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpMethod;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpRequest;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpVersion;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.QueryStringDecoder;
+
+import java.util.Map;
+
+import static java.util.Objects.requireNonNull;
+
+/**
+ * This class replaces the standard error response to be identical with those sent by the {@link AbstractRestHandler}.
+ */
+public class RouterHandler extends SimpleChannelInboundHandler<HttpRequest> {
+ public static final String ROUTER_HANDLER_NAME = RouterHandler.class.getName() + "_ROUTER_HANDLER";
+ public static final String ROUTED_HANDLER_NAME = RouterHandler.class.getName() + "_ROUTED_HANDLER";
+
+ private final Map<String, String> responseHeaders;
+ private final Router router;
+
+ public RouterHandler(Router router, final Map<String, String> responseHeaders) {
+ this.router = requireNonNull(router);
+ this.responseHeaders = requireNonNull(responseHeaders);
+ }
+
+ public String getName() {
+ return ROUTER_HANDLER_NAME;
+ }
+
+ @Override
+ protected void channelRead0(ChannelHandlerContext channelHandlerContext, HttpRequest httpRequest) throws Exception {
+ if (HttpHeaders.is100ContinueExpected(httpRequest)) {
+ channelHandlerContext.writeAndFlush(new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.CONTINUE));
+ return;
+ }
+
+ // Route
+ HttpMethod method = httpRequest.getMethod();
+ QueryStringDecoder qsd = new QueryStringDecoder(httpRequest.getUri());
+ RouteResult routeResult = router.route(method, qsd.path(), qsd.parameters());
+
+ if (routeResult == null) {
+ respondNotFound(channelHandlerContext, httpRequest);
+ return;
+ }
+
+ routed(channelHandlerContext, routeResult, httpRequest);
+ }
+
+ private void routed(
+ ChannelHandlerContext channelHandlerContext,
+ RouteResult routeResult,
+ HttpRequest httpRequest) throws Exception {
+ ChannelInboundHandler handler = (ChannelInboundHandler) routeResult.target();
+
+ // The handler may have been added (keep alive)
+ ChannelPipeline pipeline = channelHandlerContext.pipeline();
+ ChannelHandler addedHandler = pipeline.get(ROUTED_HANDLER_NAME);
+ if (handler != addedHandler) {
+ if (addedHandler == null) {
+ pipeline.addAfter(ROUTER_HANDLER_NAME, ROUTED_HANDLER_NAME, handler);
+ } else {
+ pipeline.replace(addedHandler, ROUTED_HANDLER_NAME, handler);
+ }
+ }
+
+ channelHandlerContext.fireChannelRead(new RoutedRequest(routeResult, httpRequest).retain());
--- End diff --
:+1:
---
[GitHub] flink pull request #6031: [FLINK-9386] Embed netty router
Posted by uce <gi...@git.apache.org>.
Github user uce commented on a diff in the pull request:
https://github.com/apache/flink/pull/6031#discussion_r190246117
--- Diff: flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/RuntimeMonitorHandler.java ---
@@ -89,19 +91,20 @@ public RuntimeMonitorHandler(
}
@Override
- protected void respondAsLeader(ChannelHandlerContext ctx, Routed routed, JobManagerGateway jobManagerGateway) {
+ protected void respondAsLeader(ChannelHandlerContext ctx, RoutedRequest routedRequest, JobManagerGateway jobManagerGateway) {
CompletableFuture<FullHttpResponse> responseFuture;
+ RouteResult result = routedRequest.getRouteResult();
--- End diff --
I think if you do `RouteResult<?> result = routedRequest.getRouteResult();` you don't need the cast to `Set<String>` in lines 101 and 106.
---
[GitHub] flink pull request #6031: [FLINK-9386] Embed netty router
Posted by uce <gi...@git.apache.org>.
Github user uce commented on a diff in the pull request:
https://github.com/apache/flink/pull/6031#discussion_r190191066
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/router/RouterHandler.java ---
@@ -0,0 +1,109 @@
+/*
+ * 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.apache.flink.runtime.rest.handler.AbstractRestHandler;
+import org.apache.flink.runtime.rest.handler.util.HandlerUtils;
+import org.apache.flink.runtime.rest.messages.ErrorResponseBody;
+
+import org.apache.flink.shaded.netty4.io.netty.channel.ChannelHandler;
+import org.apache.flink.shaded.netty4.io.netty.channel.ChannelHandlerContext;
+import org.apache.flink.shaded.netty4.io.netty.channel.ChannelInboundHandler;
+import org.apache.flink.shaded.netty4.io.netty.channel.ChannelPipeline;
+import org.apache.flink.shaded.netty4.io.netty.channel.SimpleChannelInboundHandler;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.DefaultFullHttpResponse;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpHeaders;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpMethod;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpRequest;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpVersion;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.QueryStringDecoder;
+
+import java.util.Map;
+
+import static java.util.Objects.requireNonNull;
+
+/**
+ * This class replaces the standard error response to be identical with those sent by the {@link AbstractRestHandler}.
+ */
+public class RouterHandler extends SimpleChannelInboundHandler<HttpRequest> {
--- End diff --
I've verified that this class merges the behaviour of `Handler` and `AbstractHandler`. 👍
Since we copied the code, we might leave it as is, but I noticed the following minor things (there are similar warnings in the other copied classes):
- `L46`, `L47`: fields can be private
- `L62`: `throws Exception` can be removed
- `L98`: I get a unchecked call warning for `RouteResult`. We could use `<?>` to get rid of it I think
---
[GitHub] flink issue #6031: [FLINK-9386] Embed netty router
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on the issue:
https://github.com/apache/flink/pull/6031
Thanks :) I have squashed commits and rebased the PR on latest master.
---
[GitHub] flink pull request #6031: [FLINK-9386] Embed netty router
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6031#discussion_r190273990
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/router/RouterHandler.java ---
@@ -0,0 +1,109 @@
+/*
+ * 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.apache.flink.runtime.rest.handler.AbstractRestHandler;
+import org.apache.flink.runtime.rest.handler.util.HandlerUtils;
+import org.apache.flink.runtime.rest.messages.ErrorResponseBody;
+
+import org.apache.flink.shaded.netty4.io.netty.channel.ChannelHandler;
+import org.apache.flink.shaded.netty4.io.netty.channel.ChannelHandlerContext;
+import org.apache.flink.shaded.netty4.io.netty.channel.ChannelInboundHandler;
+import org.apache.flink.shaded.netty4.io.netty.channel.ChannelPipeline;
+import org.apache.flink.shaded.netty4.io.netty.channel.SimpleChannelInboundHandler;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.DefaultFullHttpResponse;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpHeaders;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpMethod;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpRequest;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpVersion;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.QueryStringDecoder;
+
+import java.util.Map;
+
+import static java.util.Objects.requireNonNull;
+
+/**
+ * This class replaces the standard error response to be identical with those sent by the {@link AbstractRestHandler}.
+ */
+public class RouterHandler extends SimpleChannelInboundHandler<HttpRequest> {
+ public static final String ROUTER_HANDLER_NAME = RouterHandler.class.getName() + "_ROUTER_HANDLER";
+ public static final String ROUTED_HANDLER_NAME = RouterHandler.class.getName() + "_ROUTED_HANDLER";
+
+ private final Map<String, String> responseHeaders;
+ private final Router router;
+
+ public RouterHandler(Router router, final Map<String, String> responseHeaders) {
+ this.router = requireNonNull(router);
+ this.responseHeaders = requireNonNull(responseHeaders);
+ }
+
+ public String getName() {
+ return ROUTER_HANDLER_NAME;
+ }
+
+ @Override
+ protected void channelRead0(ChannelHandlerContext channelHandlerContext, HttpRequest httpRequest) throws Exception {
+ if (HttpHeaders.is100ContinueExpected(httpRequest)) {
+ channelHandlerContext.writeAndFlush(new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.CONTINUE));
+ return;
+ }
+
+ // Route
+ HttpMethod method = httpRequest.getMethod();
+ QueryStringDecoder qsd = new QueryStringDecoder(httpRequest.getUri());
+ RouteResult routeResult = router.route(method, qsd.path(), qsd.parameters());
+
+ if (routeResult == null) {
+ respondNotFound(channelHandlerContext, httpRequest);
+ return;
+ }
+
+ routed(channelHandlerContext, routeResult, httpRequest);
+ }
+
+ private void routed(
+ ChannelHandlerContext channelHandlerContext,
+ RouteResult routeResult,
+ HttpRequest httpRequest) throws Exception {
+ ChannelInboundHandler handler = (ChannelInboundHandler) routeResult.target();
+
+ // The handler may have been added (keep alive)
+ ChannelPipeline pipeline = channelHandlerContext.pipeline();
+ ChannelHandler addedHandler = pipeline.get(ROUTED_HANDLER_NAME);
+ if (handler != addedHandler) {
+ if (addedHandler == null) {
+ pipeline.addAfter(ROUTER_HANDLER_NAME, ROUTED_HANDLER_NAME, handler);
+ } else {
+ pipeline.replace(addedHandler, ROUTED_HANDLER_NAME, handler);
+ }
+ }
+
+ channelHandlerContext.fireChannelRead(new RoutedRequest(routeResult, httpRequest).retain());
--- End diff --
changed into
```
RoutedRequest<?> request = new RoutedRequest<>(routeResult, httpRequest);
channelHandlerContext.fireChannelRead(request.retain());
```
I like having `.retain()` calls to be in the same line where they are actually designed for, since it's kind of self documenting which line is the reason behind retaining.
---
[GitHub] flink pull request #6031: [FLINK-9386] Embed netty router
Posted by uce <gi...@git.apache.org>.
Github user uce commented on a diff in the pull request:
https://github.com/apache/flink/pull/6031#discussion_r190214988
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/router/RouterHandler.java ---
@@ -0,0 +1,109 @@
+/*
+ * 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.apache.flink.runtime.rest.handler.AbstractRestHandler;
+import org.apache.flink.runtime.rest.handler.util.HandlerUtils;
+import org.apache.flink.runtime.rest.messages.ErrorResponseBody;
+
+import org.apache.flink.shaded.netty4.io.netty.channel.ChannelHandler;
+import org.apache.flink.shaded.netty4.io.netty.channel.ChannelHandlerContext;
+import org.apache.flink.shaded.netty4.io.netty.channel.ChannelInboundHandler;
+import org.apache.flink.shaded.netty4.io.netty.channel.ChannelPipeline;
+import org.apache.flink.shaded.netty4.io.netty.channel.SimpleChannelInboundHandler;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.DefaultFullHttpResponse;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpHeaders;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpMethod;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpRequest;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpVersion;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.QueryStringDecoder;
+
+import java.util.Map;
+
+import static java.util.Objects.requireNonNull;
+
+/**
+ * This class replaces the standard error response to be identical with those sent by the {@link AbstractRestHandler}.
+ */
+public class RouterHandler extends SimpleChannelInboundHandler<HttpRequest> {
+ public static final String ROUTER_HANDLER_NAME = RouterHandler.class.getName() + "_ROUTER_HANDLER";
+ public static final String ROUTED_HANDLER_NAME = RouterHandler.class.getName() + "_ROUTED_HANDLER";
+
+ private final Map<String, String> responseHeaders;
+ private final Router router;
+
+ public RouterHandler(Router router, final Map<String, String> responseHeaders) {
+ this.router = requireNonNull(router);
+ this.responseHeaders = requireNonNull(responseHeaders);
+ }
+
+ public String getName() {
+ return ROUTER_HANDLER_NAME;
+ }
+
+ @Override
+ protected void channelRead0(ChannelHandlerContext channelHandlerContext, HttpRequest httpRequest) throws Exception {
+ if (HttpHeaders.is100ContinueExpected(httpRequest)) {
+ channelHandlerContext.writeAndFlush(new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.CONTINUE));
+ return;
+ }
+
+ // Route
+ HttpMethod method = httpRequest.getMethod();
+ QueryStringDecoder qsd = new QueryStringDecoder(httpRequest.getUri());
+ RouteResult routeResult = router.route(method, qsd.path(), qsd.parameters());
+
+ if (routeResult == null) {
+ respondNotFound(channelHandlerContext, httpRequest);
+ return;
+ }
+
+ routed(channelHandlerContext, routeResult, httpRequest);
+ }
+
+ private void routed(
+ ChannelHandlerContext channelHandlerContext,
+ RouteResult routeResult,
+ HttpRequest httpRequest) throws Exception {
+ ChannelInboundHandler handler = (ChannelInboundHandler) routeResult.target();
+
+ // The handler may have been added (keep alive)
+ ChannelPipeline pipeline = channelHandlerContext.pipeline();
+ ChannelHandler addedHandler = pipeline.get(ROUTED_HANDLER_NAME);
+ if (handler != addedHandler) {
+ if (addedHandler == null) {
+ pipeline.addAfter(ROUTER_HANDLER_NAME, ROUTED_HANDLER_NAME, handler);
+ } else {
+ pipeline.replace(addedHandler, ROUTED_HANDLER_NAME, handler);
+ }
+ }
+
+ channelHandlerContext.fireChannelRead(new RoutedRequest(routeResult, httpRequest).retain());
--- End diff --
Do you think it makes sense to move `new RoutedRequest(routeResult, httpRequest).retain()` to a separate line? I was wondering whether it is retained as in the original handler, because I missed the `retain()` here.
I find the following clearer, but feel free to ignore:
```java
RoutedRequest request = new RoutedRequest(routeResult, httpRequest);
request.retain();
channelHandlerContext.fireChannelRead(request);
```
---
[GitHub] flink issue #6031: [FLINK-9386] Embed netty router
Posted by uce <gi...@git.apache.org>.
Github user uce commented on the issue:
https://github.com/apache/flink/pull/6031
> I planned to upgrade netty and drop netty-router in one step of upgrading flink-shaded-netty. Do you think it should be split somehow?
No, if the ticket for upgrading covers that, we are all good.
👍 to merge. Thanks for taking care of this. Looking forward to finally have a recent Netty version.
---
[GitHub] flink pull request #6031: [FLINK-9386] Embed netty router
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6031#discussion_r190272895
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/router/MethodlessRouter.java ---
@@ -0,0 +1,121 @@
+/*
+ * 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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+
+/**
+ * This is adopted and simplified code from tv.cntt:netty-router library. For more information check {@link Router}.
+ *
+ * <p>Router that doesn't contain information about HTTP request methods and route
+ * matching orders.
+ */
+final class MethodlessRouter<T> {
+ private static final Logger log = LoggerFactory.getLogger(MethodlessRouter.class);
+
+ // A path pattern can only point to one target
+ private final Map<PathPattern, T> routes = new LinkedHashMap<>();
--- End diff --
I'm not convinced that embedding the sorting logic inside router is good idea. Now it's clear that order of adding methods matter and it should be handled by the caller. If we provide sorting option, it can add confusion, especially because it can not be mandatory option - our sorting based on number of parameters is just a hack that kind of works with our patterns at the moment, but there are corner cases that would have to be "sorted"/"ordered" manually. Like patterns:
```
/jobs/foo/:B
/jobs/:A/bar
```
Which one of it should be matched to url `/jobs/foo/bar`?
Regarding thread safety, this class is just simply not thread safe, and it should be the user's responsibility to handle it appropriately. On the other hand, turning it into immutable would hardcode some assumption and prevent some possible use cases and I'm not sure if that's worth the effort (it would add some substantial boiler plate code in storing intermediate builder's state, without clear benefits for the future since this code hasn't changed much for last couple of years). Don't get me wrong, the builder approach is nicer (regardless of sorting/immutability), but I'm just not sure if it's worth the extra effort.
---
[GitHub] flink pull request #6031: [FLINK-9386] Embed netty router
Posted by uce <gi...@git.apache.org>.
Github user uce commented on a diff in the pull request:
https://github.com/apache/flink/pull/6031#discussion_r190274800
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/router/MethodlessRouter.java ---
@@ -0,0 +1,121 @@
+/*
+ * 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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+
+/**
+ * This is adopted and simplified code from tv.cntt:netty-router library. For more information check {@link Router}.
+ *
+ * <p>Router that doesn't contain information about HTTP request methods and route
+ * matching orders.
+ */
+final class MethodlessRouter<T> {
+ private static final Logger log = LoggerFactory.getLogger(MethodlessRouter.class);
+
+ // A path pattern can only point to one target
+ private final Map<PathPattern, T> routes = new LinkedHashMap<>();
--- End diff --
Fine with me to not invest more time into this and keep it as is 👍
---
[GitHub] flink pull request #6031: [FLINK-9386] Embed netty router
Posted by uce <gi...@git.apache.org>.
Github user uce commented on a diff in the pull request:
https://github.com/apache/flink/pull/6031#discussion_r190213387
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/router/MethodlessRouter.java ---
@@ -0,0 +1,121 @@
+/*
+ * 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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+
+/**
+ * This is adopted and simplified code from tv.cntt:netty-router library. For more information check {@link Router}.
+ *
+ * <p>Router that doesn't contain information about HTTP request methods and route
+ * matching orders.
+ */
+final class MethodlessRouter<T> {
+ private static final Logger log = LoggerFactory.getLogger(MethodlessRouter.class);
+
+ // A path pattern can only point to one target
+ private final Map<PathPattern, T> routes = new LinkedHashMap<>();
--- End diff --
Using the `LinkedHashMap` is what preserves the order of adding handlers, right? Or is there another thing that I've missed?
---
[GitHub] flink pull request #6031: [FLINK-9386] Embed netty router
Posted by uce <gi...@git.apache.org>.
Github user uce commented on a diff in the pull request:
https://github.com/apache/flink/pull/6031#discussion_r190186599
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/router/RouterHandler.java ---
@@ -0,0 +1,109 @@
+/*
+ * 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.apache.flink.runtime.rest.handler.AbstractRestHandler;
+import org.apache.flink.runtime.rest.handler.util.HandlerUtils;
+import org.apache.flink.runtime.rest.messages.ErrorResponseBody;
+
+import org.apache.flink.shaded.netty4.io.netty.channel.ChannelHandler;
+import org.apache.flink.shaded.netty4.io.netty.channel.ChannelHandlerContext;
+import org.apache.flink.shaded.netty4.io.netty.channel.ChannelInboundHandler;
+import org.apache.flink.shaded.netty4.io.netty.channel.ChannelPipeline;
+import org.apache.flink.shaded.netty4.io.netty.channel.SimpleChannelInboundHandler;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.DefaultFullHttpResponse;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpHeaders;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpMethod;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpRequest;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpVersion;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.QueryStringDecoder;
+
+import java.util.Map;
+
+import static java.util.Objects.requireNonNull;
+
+/**
+ * This class replaces the standard error response to be identical with those sent by the {@link AbstractRestHandler}.
--- End diff --
- I would add a comment that this was copied and simplified from https://github.com/sinetja/netty-router/blob/1.10/src/main/java/io/netty/handler/codec/http/router/Handler.java and https://github.com/sinetja/netty-router/blob/1.10/src/main/java/io/netty/handler/codec/http/router/AbstractHandler.java for future reference. That can be beneficial in the future.
- I would also copy the high-level comment from that class: `Inbound handler that converts HttpRequest to RoutedRequest and passes RoutedRequest to the matched handler` as found in the original Handler class.
---
[GitHub] flink pull request #6031: [FLINK-9386] Embed netty router
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/flink/pull/6031
---
[GitHub] flink pull request #6031: [FLINK-9386] Embed netty router
Posted by uce <gi...@git.apache.org>.
Github user uce commented on a diff in the pull request:
https://github.com/apache/flink/pull/6031#discussion_r190248244
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/router/PathPattern.java ---
@@ -0,0 +1,179 @@
+/*
+ * 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.Map;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * This is adopted and simplified code from tv.cntt:netty-router library. For more information check {@link Router}.
+ *
+ * <p>The pattern can contain constants or placeholders, example:
+ * {@code constant1/:placeholder1/constant2/:*}.
+ *
+ * <p>{@code :*} is a special placeholder to catch the rest of the path
+ * (may include slashes). If exists, it must appear at the end of the path.
+ *
+ * <p>The pattern must not contain query, example:
+ * {@code constant1/constant2?foo=bar}.
+ *
+ * <p>The pattern will be broken to tokens, example:
+ * {@code ["constant1", ":variable", "constant2", ":*"]}
+ */
+final class PathPattern {
--- End diff --
Maybe add reference to https://github.com/sinetja/netty-router/blob/2.2.0/src/main/java/io/netty/handler/codec/http/router/PathPattern.java
---
[GitHub] flink pull request #6031: [FLINK-9386] Embed netty router
Posted by uce <gi...@git.apache.org>.
Github user uce commented on a diff in the pull request:
https://github.com/apache/flink/pull/6031#discussion_r190248637
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/router/Router.java ---
@@ -0,0 +1,398 @@
+/*
+ * 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.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpMethod;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.QueryStringDecoder;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+
+/**
+ * This is adopted and simplified code from tv.cntt:netty-router library. Compared to original version this one
--- End diff --
- Maybe add a link to https://github.com/sinetja/netty-router/blob/2.2.0/src/main/java/io/netty/handler/codec/http/router/Router.java
---
[GitHub] flink issue #6031: [FLINK-9386] Embed netty router
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on the issue:
https://github.com/apache/flink/pull/6031
Thanks for the review and manual checks! I either addressed your comments in fixup commit and left comment responses otherwise.
> Do we file a follow up ticket to remove netty-router from our shaded Netty? Optionally, we might consider adding an import suppression for org.apache.flink.shaded.netty4.io.netty.handler.codec.http.router.* to guard against any accidental usage until we remove the dependency.
I planned to upgrade netty and drop netty-router in one step of upgrading `flink-shaded-netty`. Do you think it should be split somehow?
> I don't know what your opinion is on cleaning up the copied classes. I'm OK with keeping everything as is, but I get a bunch of minor/trivial warnings about the code (like code visibility, unchecked calls, unused variables, etc.). I understand if we want to stay close to the copied classes.
I would be open to clean it up as we think it's best. I do not see a point in maintaining compatibility with the original sources. If you have other such comments, please feel free to write them.
---
[GitHub] flink pull request #6031: [FLINK-9386] Embed netty router
Posted by uce <gi...@git.apache.org>.
Github user uce commented on a diff in the pull request:
https://github.com/apache/flink/pull/6031#discussion_r190197601
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/router/MethodlessRouter.java ---
@@ -0,0 +1,121 @@
+/*
+ * 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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+
+/**
+ * This is adopted and simplified code from tv.cntt:netty-router library. For more information check {@link Router}.
--- End diff --
- Maybe add a link to https://github.com/sinetja/netty-router/blob/2.2.0/src/main/java/io/netty/handler/codec/http/router/MethodlessRouter.java and https://github.com/sinetja/netty-router/blob/2.2.0/src/main/java/io/netty/handler/codec/http/router/OrderlessRouter.java as a reference
---
[GitHub] flink pull request #6031: [FLINK-9386] Embed netty router
Posted by uce <gi...@git.apache.org>.
Github user uce commented on a diff in the pull request:
https://github.com/apache/flink/pull/6031#discussion_r190244366
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/router/MethodlessRouter.java ---
@@ -0,0 +1,121 @@
+/*
+ * 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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+
+/**
+ * This is adopted and simplified code from tv.cntt:netty-router library. For more information check {@link Router}.
+ *
+ * <p>Router that doesn't contain information about HTTP request methods and route
+ * matching orders.
+ */
+final class MethodlessRouter<T> {
+ private static final Logger log = LoggerFactory.getLogger(MethodlessRouter.class);
+
+ // A path pattern can only point to one target
+ private final Map<PathPattern, T> routes = new LinkedHashMap<>();
--- End diff --
Please correct me if I'm wrong but I have a question regarding memory visibility: The thread that updates this map and the Netty event loop threads are different, so there might theoretically be a memory visibility issue if routes are added after the router has been passed to the `RouterHandler`. I don't think that we do this currently, but the API theoretically allows it. I'm wondering whether it makes sense to make the routes immutable, maybe something like creating the routes with a builder:
```java
Routes routes = new RoutesBuilder()
.addGet(...)
...
.build();
Router router = new Router(routes);
```
Or use a thread-safe map here that also preserves ordering (maybe wrap using `synchronizedMap`).
---
Since we currently rely on correct registration order (e.g. `/jobs/overview` before `/jobs/:id` for correct matching), the immutable approach would allow us to include a utility in `Routes` that sorts pattern as done in `RestServerEndpoint:143`:
```java
Collections.sort(handlers,RestHandlerUrlComparator.INSTANCE);
```
---
[GitHub] flink pull request #6031: [FLINK-9386] Embed netty router
Posted by uce <gi...@git.apache.org>.
Github user uce commented on a diff in the pull request:
https://github.com/apache/flink/pull/6031#discussion_r190248429
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/router/RouteResult.java ---
@@ -0,0 +1,136 @@
+/*
+ * 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.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpMethod;
+import org.apache.flink.shaded.netty4.io.netty.util.internal.ObjectUtil;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * This is adopted and simplified code from tv.cntt:netty-router library. For more information check {@link Router}.
+ *
+ * <p>Result of calling {@link Router#route(HttpMethod, String)}.
+ */
+public class RouteResult<T> {
--- End diff --
Maybe add reference to https://github.com/sinetja/netty-router/blob/2.2.0/src/main/java/io/netty/handler/codec/http/router/RouteResult.java
---