This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch CAMEL-26137-netty-http-reply-fixes in repository https://gitbox.apache.org/repos/asf/camel.git
commit 79fa2b9412396958c967508321f289100d19c868 Author: Claus Ibsen <[email protected]> AuthorDate: Tue Nov 3 15:32:32 2026 +0100 CAMEL-26137: camel-netty-http - reply to a List body and to the request body sent back with disableStreamCache, keep the message with sync=false A List reply failed the byte[] conversion, so the consumer answered 500; it is now written as its toString(), as a Map or a POJO is. With disableStreamCache=true a route that sent the request ByteBuf back as the reply got no response, as the buffer was released twice; the response now writes a retained duplicate of it. With sync=false the camel-netty producer created an empty out message on an InOut exchange when checking the close header, so the exchange (also with netty-http) continued with an empty message. It now reads the header from the current message. Co-Authored-By: Claude Opus 5.5 <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../camel/catalog/components/netty-http.json | 2 +- .../apache/camel/catalog/docs/netty-component.adoc | 2 +- .../camel/component/netty/http/netty-http.json | 2 +- .../netty/http/DefaultNettyHttpBinding.java | 22 +++- .../component/netty/http/NettyHttpEndpoint.java | 4 +- .../netty/http/NettyHttpProducerSyncFalseTest.java | 59 +++++++++++ .../netty/http/NettyHttpReplyBodyTest.java | 115 +++++++++++++++++++++ .../camel-netty/src/main/docs/netty-component.adoc | 2 +- .../camel/component/netty/NettyProducer.java | 13 +-- .../netty/NettyProducerSyncFalseReplyTest.java | 14 +++ .../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 5 +- 11 files changed, 223 insertions(+), 17 deletions(-) diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/netty-http.json b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/netty-http.json index f80cea968801..cfab0670749c 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/netty-http.json +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/netty-http.json @@ -133,7 +133,7 @@ }, "bodyTypes": { "consumerOut": { "carriers": [ "stream" ], "javaType": "org.apache.camel.component.netty.http.NettyChannelBufferStreamCache", "dependsOn": [ "nettyHttpBinding" ], "byOption": [ { "option": "disableStreamCache", "value": "true", "carriers": [ "pojo", "stream" ] }, { "option": "protocol", "value": "proxy", "carriers": [ "pojo" ], "javaType": "io.netty.buffer.ByteBuf" } ], "description": "The request body as the stream cache of Camel, which can be read many times; empty for a request wi [...] - "consumerReply": { "carriers": [ "null", "text", "bytes", "stream", "file", "value" ], "description": "The response body: text, bytes, a stream, a file, or a Netty ByteBuf or HttpResponse; no body sends 204. A Map or a POJO is written as its toString() and a List is not written at all: marshal it first. A failed exchange sends 500 with an empty body (muteException)" }, + "consumerReply": { "carriers": [ "null", "text", "bytes", "stream", "file", "value" ], "description": "The response body: text, bytes, a stream, a file, or a Netty ByteBuf or HttpResponse; no body sends 204. A Map, a List or a POJO is written as its toString(): marshal it first. A failed exchange sends 500 with an empty body (muteException)" }, "producerIn": { "carriers": [ "null", "text", "bytes", "stream", "file" ], "description": "The request body: text, bytes, a stream or a file, or a Netty ByteBuf or HttpRequest. A Map, a List or a POJO fails: marshal it first. Without the CamelHttpMethod header, no body sends a GET and a body a POST" }, "producerOut": { "carriers": [ "bytes" ], "javaType": "byte[]", "dependsOn": [ "nettyHttpBinding", "sync" ], "byOption": [ { "option": "disableStreamCache", "value": "true", "carriers": [ "pojo", "stream" ] } ], "description": "The response body as a byte array, empty for a response without a body. With disableStreamCache the Netty ByteBuf of the response, or an InputStream for a chunked response; the ByteBuf is released when the exchange is done" } }, diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/netty-component.adoc b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/netty-component.adoc index 8a8a558565ba..f270d1ad9dd8 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/netty-component.adoc +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/netty-component.adoc @@ -168,7 +168,7 @@ consumer closes the TCP channel, so the client does not wait for a reply that ne With `useByteBuf=true` a body that cannot be converted to a `ByteBuf` fails the exchange instead of sending nothing. With `sync=false` the producer continues routing as soon as the request is written, and ignores any message the remote -server sends back. +server sends back. The message of the exchange stays as it was, also for an `InOut` exchange. ==== Using Multiple Codecs diff --git a/components/camel-netty-http/src/generated/resources/META-INF/org/apache/camel/component/netty/http/netty-http.json b/components/camel-netty-http/src/generated/resources/META-INF/org/apache/camel/component/netty/http/netty-http.json index f80cea968801..cfab0670749c 100644 --- a/components/camel-netty-http/src/generated/resources/META-INF/org/apache/camel/component/netty/http/netty-http.json +++ b/components/camel-netty-http/src/generated/resources/META-INF/org/apache/camel/component/netty/http/netty-http.json @@ -133,7 +133,7 @@ }, "bodyTypes": { "consumerOut": { "carriers": [ "stream" ], "javaType": "org.apache.camel.component.netty.http.NettyChannelBufferStreamCache", "dependsOn": [ "nettyHttpBinding" ], "byOption": [ { "option": "disableStreamCache", "value": "true", "carriers": [ "pojo", "stream" ] }, { "option": "protocol", "value": "proxy", "carriers": [ "pojo" ], "javaType": "io.netty.buffer.ByteBuf" } ], "description": "The request body as the stream cache of Camel, which can be read many times; empty for a request wi [...] - "consumerReply": { "carriers": [ "null", "text", "bytes", "stream", "file", "value" ], "description": "The response body: text, bytes, a stream, a file, or a Netty ByteBuf or HttpResponse; no body sends 204. A Map or a POJO is written as its toString() and a List is not written at all: marshal it first. A failed exchange sends 500 with an empty body (muteException)" }, + "consumerReply": { "carriers": [ "null", "text", "bytes", "stream", "file", "value" ], "description": "The response body: text, bytes, a stream, a file, or a Netty ByteBuf or HttpResponse; no body sends 204. A Map, a List or a POJO is written as its toString(): marshal it first. A failed exchange sends 500 with an empty body (muteException)" }, "producerIn": { "carriers": [ "null", "text", "bytes", "stream", "file" ], "description": "The request body: text, bytes, a stream or a file, or a Netty ByteBuf or HttpRequest. A Map, a List or a POJO fails: marshal it first. Without the CamelHttpMethod header, no body sends a GET and a body a POST" }, "producerOut": { "carriers": [ "bytes" ], "javaType": "byte[]", "dependsOn": [ "nettyHttpBinding", "sync" ], "byOption": [ { "option": "disableStreamCache", "value": "true", "carriers": [ "pojo", "stream" ] } ], "description": "The response body as a byte array, empty for a response without a body. With disableStreamCache the Netty ByteBuf of the response, or an InputStream for a chunked response; the ByteBuf is released when the exchange is done" } }, diff --git a/components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/DefaultNettyHttpBinding.java b/components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/DefaultNettyHttpBinding.java index dd73344dc665..b686fe7b77b2 100644 --- a/components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/DefaultNettyHttpBinding.java +++ b/components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/DefaultNettyHttpBinding.java @@ -47,6 +47,7 @@ import io.netty.util.ReferenceCountUtil; import org.apache.camel.Exchange; import org.apache.camel.Message; import org.apache.camel.RuntimeCamelException; +import org.apache.camel.TypeConversionException; import org.apache.camel.TypeConverter; import org.apache.camel.component.netty.NettyConverter; import org.apache.camel.http.base.HttpHeaderFilterStrategy; @@ -511,7 +512,9 @@ public class DefaultNettyHttpBinding implements NettyHttpBinding, Cloneable { if (response == null) { if (body instanceof ByteBuf byteBuf) { - buffer = byteBuf; + // the request body (with disableStreamCache) is released when the exchange is done, so a reply that + // sends it back must write a reference of its own, as writing the response releases the buffer too + buffer = isRequestBody(message, byteBuf) ? byteBuf.retainedDuplicate() : byteBuf; } else { // a String body is written in the charset of the content type buffer = toByteBufWithContentTypeCharset(message, body); @@ -521,7 +524,14 @@ public class DefaultNettyHttpBinding implements NettyHttpBinding, Cloneable { } if (buffer == null) { // fallback to byte array as last resort - byte[] data = message.getBody(byte[].class); + byte[] data; + try { + data = message.getBody(byte[].class); + } catch (TypeConversionException e) { + // such as a List, which is then written as its String form + LOG.trace("Cannot convert the body to byte[], writing it as a String: {}", e.getMessage()); + data = null; + } if (data != null) { buffer = NettyConverter.toByteBuffer(data); } else { @@ -605,6 +615,14 @@ public class DefaultNettyHttpBinding implements NettyHttpBinding, Cloneable { return response; } + /** + * Whether the buffer is the body of the HTTP request of the exchange. + */ + private static boolean isRequestBody(Message message, ByteBuf buffer) { + return message.getExchange().getIn() instanceof NettyHttpMessage in && in.getHttpRequest() != null + && in.getHttpRequest().content() == buffer; + } + @Override public HttpRequest toNettyRequest(Message message, String fullUri, NettyHttpConfiguration configuration) throws Exception { LOG.trace("toNettyRequest: {}", message); diff --git a/components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/NettyHttpEndpoint.java b/components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/NettyHttpEndpoint.java index c160c004622c..474e004e1500 100644 --- a/components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/NettyHttpEndpoint.java +++ b/components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/NettyHttpEndpoint.java @@ -71,8 +71,8 @@ import org.slf4j.LoggerFactory; Carrier.NULL, Carrier.TEXT, Carrier.BYTES, Carrier.STREAM, Carrier.FILE, Carrier.VALUE }, description = "The response body: text, bytes, a stream, a file, or a Netty ByteBuf or" - + " HttpResponse; no body sends 204. A Map or a POJO is written as its" - + " toString() and a List is not written at all: marshal it first. A" + + " HttpResponse; no body sends 204. A Map, a List or a POJO is written as" + + " its toString(): marshal it first. A" + " failed exchange sends 500 with an empty body (muteException)"), producerIn = @BodyType(value = { Carrier.NULL, Carrier.TEXT, Carrier.BYTES, Carrier.STREAM, Carrier.FILE }, description = "The request body: text, bytes, a stream or a file, or a Netty ByteBuf or" diff --git a/components/camel-netty-http/src/test/java/org/apache/camel/component/netty/http/NettyHttpProducerSyncFalseTest.java b/components/camel-netty-http/src/test/java/org/apache/camel/component/netty/http/NettyHttpProducerSyncFalseTest.java new file mode 100644 index 000000000000..2ed77ea45338 --- /dev/null +++ b/components/camel-netty-http/src/test/java/org/apache/camel/component/netty/http/NettyHttpProducerSyncFalseTest.java @@ -0,0 +1,59 @@ +/* + * 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.camel.component.netty.http; + +import java.util.concurrent.TimeUnit; + +import org.apache.camel.Exchange; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.junit.jupiter.api.Test; + +import static org.awaitility.Awaitility.await; + +/** + * With sync=false the producer does not wait for the response, and the response that arrives later is not set on the + * exchange. + */ +public class NettyHttpProducerSyncFalseTest extends BaseNettyTestSupport { + + @Test + public void testResponseIgnored() throws Exception { + MockEndpoint server = getMockEndpoint("mock:server"); + server.expectedBodiesReceived("Hello World"); + + Exchange exchange = template.request("netty-http:http://localhost:{{port}}/foo?sync=false";, + e -> e.getIn().setBody("Hello World")); + + MockEndpoint.assertIsSatisfied(context); + // the response the server sends back must not replace the message of the exchange that already completed + await().during(500, TimeUnit.MILLISECONDS).atMost(5, TimeUnit.SECONDS) + .until(() -> "Hello World".equals(exchange.getMessage().getBody(String.class))); + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + from("netty-http:http://0.0.0.0:{{port}}/foo";) + .to("mock:server") + .setBody().constant("Bye World"); + } + }; + } +} diff --git a/components/camel-netty-http/src/test/java/org/apache/camel/component/netty/http/NettyHttpReplyBodyTest.java b/components/camel-netty-http/src/test/java/org/apache/camel/component/netty/http/NettyHttpReplyBodyTest.java new file mode 100644 index 000000000000..df37e0181056 --- /dev/null +++ b/components/camel-netty-http/src/test/java/org/apache/camel/component/netty/http/NettyHttpReplyBodyTest.java @@ -0,0 +1,115 @@ +/* + * 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.camel.component.netty.http; + +import java.util.List; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import io.netty.buffer.ByteBuf; +import org.apache.camel.builder.RouteBuilder; +import org.apache.hc.client5.http.classic.methods.HttpGet; +import org.apache.hc.client5.http.classic.methods.HttpPost; +import org.apache.hc.client5.http.classic.methods.HttpUriRequestBase; +import org.apache.hc.client5.http.config.RequestConfig; +import org.apache.hc.client5.http.impl.classic.CloseableHttpClient; +import org.apache.hc.client5.http.impl.classic.CloseableHttpResponse; +import org.apache.hc.client5.http.impl.classic.HttpClients; +import org.apache.hc.core5.http.ContentType; +import org.apache.hc.core5.http.io.entity.EntityUtils; +import org.apache.hc.core5.http.io.entity.StringEntity; +import org.apache.hc.core5.util.Timeout; +import org.junit.jupiter.api.Test; + +import static org.awaitility.Awaitility.await; +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** + * The consumer writes a reply for a List body, and for the request body sent back as-is with disableStreamCache=true. + */ +public class NettyHttpReplyBodyTest extends BaseNettyTestSupport { + + private final AtomicReference<ByteBuf> requestBody = new AtomicReference<>(); + + @Test + public void testListReply() throws Exception { + Reply reply = execute(new HttpGet("http://localhost:"; + getPort() + "/list")); + assertEquals(200, reply.code()); + assertEquals("[a, b]", reply.body()); + } + + @Test + public void testEchoRequestBodyWithDisableStreamCache() throws Exception { + for (int i = 0; i < 3; i++) { + Reply reply = execute(post("/echo", "Hello World " + i)); + assertEquals(200, reply.code()); + assertEquals("Hello World " + i, reply.body()); + } + assertRequestBodyReleased(); + } + + @Test + public void testEchoRequestBodyFromAnotherThreadWithDisableStreamCache() throws Exception { + for (int i = 0; i < 3; i++) { + Reply reply = execute(post("/echoThreads", "Hello World " + i)); + assertEquals(200, reply.code()); + assertEquals("Hello World " + i, reply.body()); + } + assertRequestBodyReleased(); + } + + private void assertRequestBodyReleased() { + // the reply writes its own reference of the request buffer, so the buffer is still released once + await().atMost(10, TimeUnit.SECONDS).until(() -> requestBody.get().refCnt() == 0); + } + + private HttpPost post(String path, String body) { + HttpPost post = new HttpPost("http://localhost:"; + getPort() + path); + post.setEntity(new StringEntity(body, ContentType.TEXT_PLAIN)); + return post; + } + + private Reply execute(HttpUriRequestBase request) throws Exception { + // before, the consumer sent no response at all, so do not wait for one forever + RequestConfig config = RequestConfig.custom().setResponseTimeout(Timeout.of(10, TimeUnit.SECONDS)).build(); + try (CloseableHttpClient client = HttpClients.custom().setDefaultRequestConfig(config).build(); + CloseableHttpResponse response = client.execute(request)) { + return new Reply(response.getCode(), EntityUtils.toString(response.getEntity(), "UTF-8")); + } + } + + private record Reply(int code, String body) { + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + from("netty-http:http://0.0.0.0:{{port}}/list";) + .process(exchange -> exchange.getMessage().setBody(List.of("a", "b"))); + + from("netty-http:http://0.0.0.0:{{port}}/echo?disableStreamCache=true";) + .process(exchange -> requestBody.set(exchange.getMessage().getBody(ByteBuf.class))); + + from("netty-http:http://0.0.0.0:{{port}}/echoThreads?disableStreamCache=true";) + .process(exchange -> requestBody.set(exchange.getMessage().getBody(ByteBuf.class))) + .threads(1); + } + }; + } +} diff --git a/components/camel-netty/src/main/docs/netty-component.adoc b/components/camel-netty/src/main/docs/netty-component.adoc index 8a8a558565ba..f270d1ad9dd8 100644 --- a/components/camel-netty/src/main/docs/netty-component.adoc +++ b/components/camel-netty/src/main/docs/netty-component.adoc @@ -168,7 +168,7 @@ consumer closes the TCP channel, so the client does not wait for a reply that ne With `useByteBuf=true` a body that cannot be converted to a `ByteBuf` fails the exchange instead of sending nothing. With `sync=false` the producer continues routing as soon as the request is written, and ignores any message the remote -server sends back. +server sends back. The message of the exchange stays as it was, also for an `InOut` exchange. ==== Using Multiple Codecs diff --git a/components/camel-netty/src/main/java/org/apache/camel/component/netty/NettyProducer.java b/components/camel-netty/src/main/java/org/apache/camel/component/netty/NettyProducer.java index af53bd2a532f..c6dfcfac0114 100644 --- a/components/camel-netty/src/main/java/org/apache/camel/component/netty/NettyProducer.java +++ b/components/camel-netty/src/main/java/org/apache/camel/component/netty/NettyProducer.java @@ -390,14 +390,11 @@ public class NettyProducer extends DefaultAsyncProducer { // if we do not expect any reply then signal callback to continue routing if (!configuration.isSync()) { try { - // should channel be closed after complete? - Boolean close; - if (ExchangeHelper.isOutCapable(exchange)) { - close = exchange.getOut().getHeader(NettyConstants.NETTY_CLOSE_CHANNEL_WHEN_COMPLETE, - Boolean.class); - } else { - close = exchange.getIn().getHeader(NettyConstants.NETTY_CLOSE_CHANNEL_WHEN_COMPLETE, Boolean.class); - } + // should channel be closed after complete? (no reply is received, so do not create an empty + // out message on an InOut exchange, which would replace the message the exchange continues with) + Boolean close + = exchange.getMessage().getHeader(NettyConstants.NETTY_CLOSE_CHANNEL_WHEN_COMPLETE, + Boolean.class); // should we disconnect, the header can override the configuration boolean disconnect = getConfiguration().isDisconnect(); diff --git a/components/camel-netty/src/test/java/org/apache/camel/component/netty/NettyProducerSyncFalseReplyTest.java b/components/camel-netty/src/test/java/org/apache/camel/component/netty/NettyProducerSyncFalseReplyTest.java index b463a6cd0f2b..c4dade508389 100644 --- a/components/camel-netty/src/test/java/org/apache/camel/component/netty/NettyProducerSyncFalseReplyTest.java +++ b/components/camel-netty/src/test/java/org/apache/camel/component/netty/NettyProducerSyncFalseReplyTest.java @@ -23,6 +23,7 @@ import org.apache.camel.component.mock.MockEndpoint; import org.junit.jupiter.api.Test; import static org.awaitility.Awaitility.await; +import static org.junit.jupiter.api.Assertions.assertEquals; /** * With sync=false the producer continues routing as soon as the request is written, so a reply that the remote server @@ -41,6 +42,19 @@ class NettyProducerSyncFalseReplyTest extends BaseNettyTest { MockEndpoint.assertIsSatisfied(context); } + @Test + void anInOutExchangeKeepsItsMessage() throws Exception { + getMockEndpoint("mock:server").expectedBodiesReceived("Hello World"); + getMockEndpoint("mock:result").expectedBodiesReceived("Hello World"); + getMockEndpoint("mock:result").setAssertPeriod(500); + + // no reply is received, so the producer must not leave an empty out message on an InOut exchange + Object reply = template.requestBody("direct:start", "Hello World"); + + MockEndpoint.assertIsSatisfied(context); + assertEquals("Hello World", reply); + } + @Override protected RouteBuilder createRouteBuilder() { return new RouteBuilder() { diff --git a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc index 7c82ebd18930..85b0596c68d1 100644 --- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc +++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc @@ -2209,7 +2209,8 @@ also converts a reply set as the out message to a `ByteBuf`. `NettyPayloadHelper `InvalidPayloadException`. With `sync=false` the producer now ignores a message the remote server sends back. Before, that reply was set on the -exchange after it had already continued routing. +exchange after it had already continued routing. The producer now also leaves the body and headers of an `InOut` +exchange as they were. Before, the exchange continued with an empty message. This also applies to `netty-http`. === camel-netty - a reply that cannot be written, and allowSerializedHeaders on the reply @@ -2243,6 +2244,8 @@ When no HTTP response can be built from the reply of the route (for example the now answers with HTTP Status 500 and closes the connection. The body is empty, or the stack trace of the exception with `muteException=false`. Before, no response was sent and the client waited until it timed out. +A `List` reply is now written as its `toString()`, as a `Map` or a POJO already was. Before, no response was sent. + === camel-odata - the body of CREATE and UPDATE, and a response that is not a JSON object `CREATE` and `UPDATE` send JSON text given as a byte array, a stream or a file as it is, as they do for a `String`.
