@Override
    protected void respondAsLeader(
            ChannelHandlerContext ctx, RoutedRequest routedRequest, T gateway) {
        HttpRequest httpRequest = routedRequest.getRequest();
        if (log.isTraceEnabled()) {
            log.trace("Received request " + httpRequest.uri() + '.');
        }

        FileUploads uploadedFiles = null;
        try {
            if (!inFlightRequestTracker.registerRequest()) {
                log.debug(
                        "The handler instance for {} had already been closed.",
                        untypedResponseMessageHeaders.getTargetRestEndpointURL());
                ctx.channel().close();
                return;
            }

            if (!(httpRequest instanceof FullHttpRequest)) {
                // The RestServerEndpoint defines a HttpObjectAggregator in the pipeline that always
                // returns
                // FullHttpRequests.
                log.error(
                        "Implementation error: Received a request that wasn't a FullHttpRequest.");
                throw new RestHandlerException(
                        "Bad request received.", HttpResponseStatus.BAD_REQUEST);
            }

            final ByteBuf msgContent = ((FullHttpRequest) httpRequest).content();

            uploadedFiles = FileUploadHandler.getMultipartFileUploads(ctx);

            if (!untypedResponseMessageHeaders.acceptsFileUploads()
                    && !uploadedFiles.getUploadedFiles().isEmpty()) {
                throw new RestHandlerException(
                        "File uploads not allowed.", HttpResponseStatus.BAD_REQUEST);
            }

            R request;
            if (msgContent.capacity() == 0) {
                try {
                    request =
                            MAPPER.readValue("{}", untypedResponseMessageHeaders.getRequestClass());
                } catch (JsonParseException | JsonMappingException je) {
                    throw new RestHandlerException(
                            "Bad request received. Request did not conform to expected format.",
                            HttpResponseStatus.BAD_REQUEST,
                            je);
                }
            } else {
                try {
                    InputStream in = new ByteBufInputStream(msgContent);
                    request = MAPPER.readValue(in, untypedResponseMessageHeaders.getRequestClass());
                } catch (JsonParseException | JsonMappingException je) {
                    throw new RestHandlerException(
                            String.format(
                                    "Request did not match expected format %s.",
                                    untypedResponseMessageHeaders
                                            .getRequestClass()
                                            .getSimpleName()),
                            HttpResponseStatus.BAD_REQUEST,
                            je);
                }
            }

            final HandlerRequest<R> handlerRequest;

            try {
                handlerRequest =
                        HandlerRequest.resolveParametersAndCreate(
                                request,
                                untypedResponseMessageHeaders.getUnresolvedMessageParameters(),
                                routedRequest.getRouteResult().pathParams(),
                                routedRequest.getRouteResult().queryParams(),
                                uploadedFiles.getUploadedFiles());
            } catch (HandlerRequestException hre) {
                log.error("Could not create the handler request.", hre);
                throw new RestHandlerException(
                        String.format(
                                "Bad request, could not parse parameters: %s", hre.getMessage()),
                        HttpResponseStatus.BAD_REQUEST,
                        hre);
            }

            log.trace("Starting request processing.");
            CompletableFuture<Void> requestProcessingFuture =
                    respondToRequest(ctx, httpRequest, handlerRequest, gateway);

            final FileUploads finalUploadedFiles = uploadedFiles;
            requestProcessingFuture
                    .handle(
                            (Void ignored, Throwable throwable) -> {
                                if (throwable != null) {
                                    return handleException(
                                            ExceptionUtils.stripCompletionException(throwable),
                                            ctx,
                                            httpRequest);
                                }
                                return CompletableFuture.<Void>completedFuture(null);
                            })
                    .thenCompose(Function.identity())
                    .whenComplete(
                            (Void ignored, Throwable throwable) -> {
                                if (throwable != null) {
                                    log.warn(
                                            "An exception occurred while handling another exception.",
                                            throwable);
                                }
                                finalizeRequestProcessing(finalUploadedFiles);
                            });
        } catch (Throwable e) {
            final FileUploads finalUploadedFiles = uploadedFiles;
            handleException(e, ctx, httpRequest)
                    .whenComplete(
                            (Void ignored, Throwable throwable) ->
                                    finalizeRequestProcessing(finalUploadedFiles));
        }
    }
@Test
    void testIgnoringUnknownFields() {
        RestfulGateway mockRestfulGateway = new TestingRestfulGateway.Builder().build();

        CompletableFuture<Void> requestProcessingCompleteFuture = new CompletableFuture<>();
        TestHandler handler =
                new TestHandler(requestProcessingCompleteFuture, mockGatewayRetriever);

        RouteResult<?> routeResult =
                new RouteResult<>("", "", Collections.emptyMap(), Collections.emptyMap(), "");
        String requestBody = "{\"unknown_field_should_be_ignore\": true}";
        HttpRequest request =
                new DefaultFullHttpRequest(
                        HttpVersion.HTTP_1_1,
                        HttpMethod.POST,
                        TestHandler.TestHeaders.INSTANCE.getTargetRestEndpointURL(),
                        Unpooled.wrappedBuffer(requestBody.getBytes(StandardCharsets.UTF_8)));
        RoutedRequest<?> routerRequest = new RoutedRequest<>(routeResult, request);

        AtomicReference<HttpResponse> response = new AtomicReference<>();

        FlinkHttpObjectAggregator aggregator =
                new FlinkHttpObjectAggregator(
                        RestOptions.SERVER_MAX_CONTENT_LENGTH.defaultValue(),
                        Collections.emptyMap());
        ChannelPipeline pipeline = mock(ChannelPipeline.class);
        when(pipeline.get(eq(FlinkHttpObjectAggregator.class))).thenReturn(aggregator);

        ChannelFuture succeededFuture = mock(ChannelFuture.class);
        when(succeededFuture.isSuccess()).thenReturn(true);

        Attribute<FileUploads> attribute = new SimpleAttribute();
        Channel channel = mock(Channel.class);
        when(channel.attr(any(AttributeKey.class))).thenReturn(attribute);

        ChannelHandlerContext context = mock(ChannelHandlerContext.class);
        when(context.pipeline()).thenReturn(pipeline);
        when(context.channel()).thenReturn(channel);
        when(context.write(any()))
                .thenAnswer(
                        invocation -> {
                            if (invocation.getArguments().length > 0
                                    && invocation.getArgument(0) instanceof HttpResponse) {
                                response.set(invocation.getArgument(0));
                            }
                            return succeededFuture;
                        });
        when(context.writeAndFlush(any())).thenReturn(succeededFuture);

        handler.respondAsLeader(context, routerRequest, mockRestfulGateway);
        assertThat(
                        response.get() == null
                                || response.get().status().codeClass() == HttpStatusClass.SUCCESS)
                .isTrue();
    }