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):
The response should be written from Netty's event loop thread, which is enforced by Netty if called from another thread. 
However, when enforced by Netty, sometimes the buffer won't be fully written if the server closes right after writing the buffer, triggering intermittent failures in `RestServerEndpointITCase#testShouldWaitForHandlersWhenClosing`

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;
}