Repository: apache/flink
File: flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/AbstractRestHandler.java (Java)

Code before:
        return messageHeaders;
    }

    @Override
    protected CompletableFuture<Void> respondToRequest(
            ChannelHandlerContext ctx,
            HttpRequest httpRequest,
            HandlerRequest<R> handlerRequest,
            T gateway) {
        CompletableFuture<P> response;

        try {
            response = handleRequest(handlerRequest, gateway);
        } catch (RestHandlerException e) {
            response = FutureUtils.completedExceptionally(e);
        }

        return response.thenAcceptAsync(
                resp ->
                        HandlerUtils.sendResponse(
                                ctx,
                                httpRequest,
                                resp,
                                messageHeaders.getResponseStatusCode(),
                                responseHeaders),
                ctx.executor());
    }

    /**
     * This method is called for every incoming request and returns a {@link CompletableFuture}
     * containing a the response.
     *
     * <p>Implementations may decide whether to throw {@link RestHandlerException}s or fail the
     * returned {@link CompletableFuture} with a {@link RestHandlerException}.
     *
     * <p>Failing the future with another exception type or throwing unchecked exceptions is
     * regarded as an implementation error as it does not allow us to provide a meaningful HTTP
     * status code. In this case a {@link HttpResponseStatus#INTERNAL_SERVER_ERROR} will be
     * returned.
     *
     * @param request request that should be handled
     * @param gateway leader gateway
     * @return future containing a handler response
     * @throws RestHandlerException if the handling failed
     */
    protected abstract CompletableFuture<P> handleRequest(
            @Nonnull HandlerRequest<R> request, @Nonnull T gateway) throws RestHandlerException;
}


Review comment (bug):
<del>@snuyanzin the issue was this stuck thread:</del>
```
Jan 14 21:15:14 	at org.apache.flink.core.testutils.BlockerSync.awaitBlocker(BlockerSync.java:59)
Jan 14 21:15:14 	- locked <0x00000000aa141c48> (a java.lang.Object)
Jan 14 21:15:14 	at org.apache.flink.runtime.rest.RestServerEndpointITCase.testShouldWaitForHandlersWhenClosing(RestServerEndpointITCase.java:597)
```

<del>Something has changed internally in Netty, and this test consistently times out on Azure CI. The issue arises when the response is not written from the event loop, and Netty enqueues the write operations on the event loop instead, and then we close the REST server itself (which is the point of the test), the writes don't always go through properly.</del>

<del>By running the write operations on the event loop, this issue goes away. I think it's better to do it this way anyways, so that Netty does not have to create 3 write tasks to eventually execute on the event loop thread.</del>


Nevermind, unfortunately this only solved the issue on my own computer, but the CI still failed at the exact same steps. I'll continue digging...

Code after:
        return messageHeaders;
    }

    @Override
    protected CompletableFuture<Void> respondToRequest(
            ChannelHandlerContext ctx,
            HttpRequest httpRequest,
            HandlerRequest<R> handlerRequest,
            T gateway) {
        CompletableFuture<P> response;

        try {
            response = handleRequest(handlerRequest, gateway);
        } catch (RestHandlerException e) {
            response = FutureUtils.completedExceptionally(e);
        }

        return response.thenAccept(
                resp ->
                        HandlerUtils.sendResponse(
                                ctx,
                                httpRequest,
                                resp,
                                messageHeaders.getResponseStatusCode(),
                                responseHeaders));
    }

    /**
     * This method is called for every incoming request and returns a {@link CompletableFuture}
     * containing a the response.
     *
     * <p>Implementations may decide whether to throw {@link RestHandlerException}s or fail the
     * returned {@link CompletableFuture} with a {@link RestHandlerException}.
     *
     * <p>Failing the future with another exception type or throwing unchecked exceptions is
     * regarded as an implementation error as it does not allow us to provide a meaningful HTTP
     * status code. In this case a {@link HttpResponseStatus#INTERNAL_SERVER_ERROR} will be
     * returned.
     *
     * @param request request that should be handled
     * @param gateway leader gateway
     * @return future containing a handler response
     * @throws RestHandlerException if the handling failed
     */
    protected abstract CompletableFuture<P> handleRequest(
            @Nonnull HandlerRequest<R> request, @Nonnull T gateway) throws RestHandlerException;
}