This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new c3ce839be05f CAMEL-26136: camel-netty - convert replies for the
default codec, fail on useByteBuf, ignore replies with sync=false (#28914)
c3ce839be05f is described below
commit c3ce839be05fb8f26a2683eb8bc775e8029b7cf6
Author: Claus Ibsen <[email protected]>
AuthorDate: Mon Nov 2 20:43:36 2026 +0100
CAMEL-26136: camel-netty - convert replies for the default codec, fail on
useByteBuf, ignore replies with sync=false (#28914)
…l on useByteBuf, ignore replies with sync=false
With the default codec (String encoder only) a consumer reply that was not a
String, including the exception of a failed exchange, could not be written
and
the client waited. The consumer now converts such a reply to a String, as
textline does, and closes the TCP channel when no reply can be written.
useByteBuf=true used the non-mandatory getBody(ByteBuf.class), so a body
that
could not be converted was skipped silently; it now fails the exchange, and
the consumer reply set as out message honours useByteBuf as well.
With sync=false the producer set a late reply of the server on an exchange
that
had already continued routing (and called its callback a second time); such
messages are now ignored.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
Signed-off-by: Claus Ibsen <[email protected]>
---
.../apache/camel/catalog/docs/netty-component.adoc | 10 +++
.../camel-netty/src/main/docs/netty-component.adoc | 10 +++
.../camel/component/netty/NettyConfiguration.java | 10 +++
.../camel/component/netty/NettyPayloadHelper.java | 25 +++---
.../netty/handlers/ClientChannelHandler.java | 9 +++
.../netty/handlers/ServerChannelHandler.java | 44 +++++++++--
.../netty/NettyDefaultCodecReplyTest.java | 91 ++++++++++++++++++++++
.../netty/NettyProducerSyncFalseReplyTest.java | 64 +++++++++++++++
.../camel/component/netty/NettyUseByteBufTest.java | 86 ++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 16 ++++
10 files changed, 348 insertions(+), 17 deletions(-)
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 5ce184a15f31..8a8a558565ba 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
@@ -160,6 +160,16 @@ exception thrown by the route, so the reply carries
neither its class name nor i
client receives the line `java.lang.Exception: Exchange processing failed`.
Set `muteException=false` to write the
exception of the route, as in earlier versions.
+The default codec (no `textline`, no custom `encoders` or `decoders`) only
writes text. The consumer therefore converts
+a reply that is not a `String` (or a `ByteBuf`), such as bytes, a stream, a
file, a `Map` or the exception of a failed
+exchange, to a `String` before writing it, as it does with the textline codec.
If no reply can be written, the
+consumer closes the TCP channel, so the client does not wait for a reply that
never comes.
+
+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.
+
==== Using Multiple Codecs
In certain cases, it may be necessary to add chains of encoders and
diff --git a/components/camel-netty/src/main/docs/netty-component.adoc
b/components/camel-netty/src/main/docs/netty-component.adoc
index 5ce184a15f31..8a8a558565ba 100644
--- a/components/camel-netty/src/main/docs/netty-component.adoc
+++ b/components/camel-netty/src/main/docs/netty-component.adoc
@@ -160,6 +160,16 @@ exception thrown by the route, so the reply carries
neither its class name nor i
client receives the line `java.lang.Exception: Exchange processing failed`.
Set `muteException=false` to write the
exception of the route, as in earlier versions.
+The default codec (no `textline`, no custom `encoders` or `decoders`) only
writes text. The consumer therefore converts
+a reply that is not a `String` (or a `ByteBuf`), such as bytes, a stream, a
file, a `Map` or the exception of a failed
+exchange, to a `String` before writing it, as it does with the textline codec.
If no reply can be written, the
+consumer closes the TCP channel, so the client does not wait for a reply that
never comes.
+
+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.
+
==== Using Multiple Codecs
In certain cases, it may be necessary to add chains of encoders and
diff --git
a/components/camel-netty/src/main/java/org/apache/camel/component/netty/NettyConfiguration.java
b/components/camel-netty/src/main/java/org/apache/camel/component/netty/NettyConfiguration.java
index c08b36a4e1b0..0533043928d9 100644
---
a/components/camel-netty/src/main/java/org/apache/camel/component/netty/NettyConfiguration.java
+++
b/components/camel-netty/src/main/java/org/apache/camel/component/netty/NettyConfiguration.java
@@ -52,6 +52,7 @@ public class NettyConfiguration extends
NettyServerBootstrapConfiguration implem
private transient List<ChannelHandler> encodersList = new ArrayList<>();
private transient List<ChannelHandler> decodersList = new ArrayList<>();
+ private transient boolean defaultStringCodec;
@UriParam(label = "producer",
description = "Allows to use a timeout for the Netty producer
when calling a remote server. By default no timeout is in use. The"
@@ -343,6 +344,7 @@ public class NettyConfiguration extends
NettyServerBootstrapConfiguration implem
Charset charset = getEncoding() != null ?
Charset.forName(getEncoding()) : CharsetUtil.UTF_8;
encodersList.add(ChannelHandlerFactories.newStringEncoder(charset, protocol));
decodersList.add(ChannelHandlerFactories.newStringDecoder(charset, protocol));
+ defaultStringCodec = true;
}
if ("udp".equalsIgnoreCase(protocol)) {
decodersList.add(ChannelHandlerFactories.newDatagramPacketDecoder());
@@ -493,6 +495,14 @@ public class NettyConfiguration extends
NettyServerBootstrapConfiguration implem
return decoders;
}
+ /**
+ * Whether the default codec, which only writes a String, is in use: no
textline, no custom encoders or decoders,
+ * and allowDefaultCodec enabled.
+ */
+ public boolean isDefaultStringCodec() {
+ return defaultStringCodec;
+ }
+
public List<ChannelHandler> getEncodersAsList() {
return encodersList;
}
diff --git
a/components/camel-netty/src/main/java/org/apache/camel/component/netty/NettyPayloadHelper.java
b/components/camel-netty/src/main/java/org/apache/camel/component/netty/NettyPayloadHelper.java
index d09c5d8ae4d5..c5211e56f4d0 100644
---
a/components/camel-netty/src/main/java/org/apache/camel/component/netty/NettyPayloadHelper.java
+++
b/components/camel-netty/src/main/java/org/apache/camel/component/netty/NettyPayloadHelper.java
@@ -21,6 +21,8 @@ import java.net.InetSocketAddress;
import io.netty.buffer.ByteBuf;
import io.netty.channel.AddressedEnvelope;
import org.apache.camel.Exchange;
+import org.apache.camel.InvalidPayloadException;
+import org.apache.camel.Message;
import org.apache.camel.support.DefaultExchangeHolder;
/**
@@ -36,28 +38,31 @@ public final class NettyPayloadHelper {
//Helper class
}
- public static Object getIn(NettyEndpoint endpoint, Exchange exchange) {
+ public static Object getIn(NettyEndpoint endpoint, Exchange exchange)
throws InvalidPayloadException {
if (endpoint.getConfiguration().isTransferExchange()) {
// we should transfer the entire exchange over the wire (includes
in/out)
return DefaultExchangeHolder.marshal(exchange, true,
endpoint.getConfiguration().isAllowSerializedHeaders());
} else {
- if (endpoint.getConfiguration().isUseByteBuf()) {
- // Just leverage the type converter
- return exchange.getIn().getBody(ByteBuf.class);
- } else {
- // normal transfer using the body only
- return exchange.getIn().getBody();
- }
+ return getBody(endpoint, exchange.getIn());
}
}
- public static Object getOut(NettyEndpoint endpoint, Exchange exchange) {
+ public static Object getOut(NettyEndpoint endpoint, Exchange exchange)
throws InvalidPayloadException {
if (endpoint.getConfiguration().isTransferExchange()) {
// we should transfer the entire exchange over the wire (includes
in/out)
return DefaultExchangeHolder.marshal(exchange);
+ } else {
+ return getBody(endpoint, exchange.getOut());
+ }
+ }
+
+ private static Object getBody(NettyEndpoint endpoint, Message message)
throws InvalidPayloadException {
+ if (endpoint.getConfiguration().isUseByteBuf() && message.getBody() !=
null) {
+ // Just leverage the type converter, and fail if the body cannot
be converted instead of sending nothing
+ return message.getMandatoryBody(ByteBuf.class);
} else {
// normal transfer using the body only
- return exchange.getOut().getBody();
+ return message.getBody();
}
}
diff --git
a/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ClientChannelHandler.java
b/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ClientChannelHandler.java
index 2205e096f57e..287aa046cfe0 100644
---
a/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ClientChannelHandler.java
+++
b/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ClientChannelHandler.java
@@ -156,6 +156,15 @@ public class ClientChannelHandler extends
SimpleChannelInboundHandler<Object> {
LOG.trace("Message received: {}", msg);
}
+ if (!producer.getConfiguration().isSync()) {
+ // the exchange continued routing as soon as the request was
written, so a message the remote server
+ // sends back must not be set on it (or on the next exchange that
uses this channel)
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Ignoring message received on channel: {} as the
producer is not sync", ctx.channel());
+ }
+ return;
+ }
+
ChannelHandler handler = ctx.pipeline().get("timeout");
if (handler != null) {
LOG.trace("Removing timeout channel as we received message");
diff --git
a/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ServerChannelHandler.java
b/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ServerChannelHandler.java
index ba78e535ad1e..54ba8f50caa2 100644
---
a/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ServerChannelHandler.java
+++
b/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ServerChannelHandler.java
@@ -18,12 +18,14 @@ package org.apache.camel.component.netty.handlers;
import java.net.SocketAddress;
+import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import org.apache.camel.Exchange;
import org.apache.camel.ExchangePattern;
import org.apache.camel.ExchangePropertyKey;
+import org.apache.camel.component.netty.NettyConfiguration;
import org.apache.camel.component.netty.NettyConstants;
import org.apache.camel.component.netty.NettyConsumer;
import org.apache.camel.component.netty.NettyHelper;
@@ -179,7 +181,23 @@ public class ServerChannelHandler extends
SimpleChannelInboundHandler<Object> {
}
private void sendResponse(ChannelHandlerContext ctx, Exchange exchange)
throws Exception {
- Object body = getResponseBody(exchange);
+ Object body;
+ try {
+ body = getResponseBody(exchange);
+ if (body != null) {
+ body = convertResponseBody(exchange, body);
+ }
+ } catch (Exception e) {
+ if (consumer.getConfiguration().isTcp()) {
+ // no reply can be written, so close the channel, otherwise
the client waits for a reply forever
+ // (if not timing out)
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Closing channel as the reply cannot be written
at address: {}", ctx.channel().remoteAddress());
+ }
+ NettyHelper.close(ctx.channel());
+ }
+ throw e;
+ }
if (body == null) {
noReplyLogger.log("No payload to send as reply for exchange: " +
exchange);
@@ -192,12 +210,6 @@ public class ServerChannelHandler extends
SimpleChannelInboundHandler<Object> {
NettyHelper.close(ctx.channel());
}
} else {
- // if textline enabled then covert to a String which must be used
for textline
- if (consumer.getConfiguration().isTextline()) {
- body = NettyHelper.getTextlineBody(body, exchange,
consumer.getConfiguration().getDelimiter(),
- consumer.getConfiguration().isAutoAppendDelimiter());
- }
-
// we got a body to write
ChannelFutureListener listener =
createResponseFutureListener(consumer, exchange, ctx.channel().remoteAddress());
if (consumer.getConfiguration().isTcp()) {
@@ -210,6 +222,24 @@ public class ServerChannelHandler extends
SimpleChannelInboundHandler<Object> {
}
}
+ private Object convertResponseBody(Exchange exchange, Object body) throws
Exception {
+ NettyConfiguration configuration = consumer.getConfiguration();
+ // if textline enabled then covert to a String which must be used for
textline
+ if (configuration.isTextline()) {
+ return NettyHelper.getTextlineBody(body, exchange,
configuration.getDelimiter(),
+ configuration.isAutoAppendDelimiter());
+ }
+ // the default codec only writes a String (or a ByteBuf as-is), so
convert the reply to a String as textline does,
+ // otherwise the write fails and the client gets no reply. Only the
pipeline of camel-netty itself has the
+ // default codec, not one built by a custom ServerInitializerFactory.
+ if (configuration.isDefaultStringCodec() &&
configuration.getServerInitializerFactory() == null
+ && !configuration.isTransferExchange()
+ && !(body instanceof CharSequence) && !(body instanceof
ByteBuf)) {
+ return
exchange.getContext().getTypeConverter().mandatoryConvertTo(String.class,
exchange, body);
+ }
+ return body;
+ }
+
/**
* Gets the object we want to use as the response object for sending to
netty.
*
diff --git
a/components/camel-netty/src/test/java/org/apache/camel/component/netty/NettyDefaultCodecReplyTest.java
b/components/camel-netty/src/test/java/org/apache/camel/component/netty/NettyDefaultCodecReplyTest.java
new file mode 100644
index 000000000000..b2489e8d8a6b
--- /dev/null
+++
b/components/camel-netty/src/test/java/org/apache/camel/component/netty/NettyDefaultCodecReplyTest.java
@@ -0,0 +1,91 @@
+/*
+ * 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;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Map;
+
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.test.AvailablePortFinder;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * The default codec (no textline, no custom encoders) only writes a String: a
reply of another type, including the
+ * exception of a failed exchange, is converted to a String instead of failing
the write and leaving the client waiting.
+ */
+class NettyDefaultCodecReplyTest extends BaseNettyTest {
+
+ private static final String DETAIL =
"the-internal-detail-a-peer-must-not-see";
+
+ @RegisterExtension
+ AvailablePortFinder.Port unmutedPort = AvailablePortFinder.find();
+
+ @RegisterExtension
+ AvailablePortFinder.Port mapPort = AvailablePortFinder.find();
+
+ @RegisterExtension
+ AvailablePortFinder.Port bytesPort = AvailablePortFinder.find();
+
+ @Test
+ void aFailedExchangeRepliesTheMutedException() {
+ assertEquals("java.lang.Exception: Exchange processing failed",
request(port.getPort()));
+ }
+
+ @Test
+ void aFailedExchangeRepliesTheExceptionWhenNotMuted() {
+ assertEquals("java.lang.IllegalStateException: " + DETAIL,
request(unmutedPort.getPort()));
+ }
+
+ @Test
+ void aMapReplyIsConvertedToString() {
+ assertEquals("{greeting=Bye World}", request(mapPort.getPort()));
+ }
+
+ @Test
+ void aByteArrayReplyIsConvertedToString() {
+ assertEquals("Bye World", request(bytesPort.getPort()));
+ }
+
+ private String request(int port) {
+ // without the conversion the client gets no reply and fails with a
ReadTimeoutException
+ return template.requestBody("netty:tcp://localhost:" + port +
"?sync=true&requestTimeout=5000", "Hello World",
+ String.class);
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("netty:tcp://localhost:{{port}}?sync=true")
+ .throwException(new IllegalStateException(DETAIL));
+
+
fromF("netty:tcp://localhost:%s?sync=true&muteException=false",
unmutedPort.getPort())
+ .throwException(new IllegalStateException(DETAIL));
+
+ fromF("netty:tcp://localhost:%s?sync=true", mapPort.getPort())
+ .process(exchange ->
exchange.getMessage().setBody(Map.of("greeting", "Bye World")));
+
+ fromF("netty:tcp://localhost:%s?sync=true",
bytesPort.getPort())
+ .process(exchange ->
exchange.getMessage().setBody("Bye World".getBytes(StandardCharsets.UTF_8)));
+ }
+ };
+ }
+}
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
new file mode 100644
index 000000000000..b463a6cd0f2b
--- /dev/null
+++
b/components/camel-netty/src/test/java/org/apache/camel/component/netty/NettyProducerSyncFalseReplyTest.java
@@ -0,0 +1,64 @@
+/*
+ * 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;
+
+import java.util.concurrent.TimeUnit;
+
+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 continues routing as soon as the request is
written, so a reply that the remote server
+ * sends later must not be set on the exchange.
+ */
+class NettyProducerSyncFalseReplyTest extends BaseNettyTest {
+
+ @Test
+ void theReplyOfTheServerIsNotSetOnTheExchange() throws Exception {
+ getMockEndpoint("mock:server").expectedBodiesReceived("Hello World");
+ getMockEndpoint("mock:result").expectedBodiesReceived("Hello World");
+ getMockEndpoint("mock:result").setAssertPeriod(500);
+
+ template.sendBody("direct:start", "Hello World");
+
+ MockEndpoint.assertIsSatisfied(context);
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start")
+ .to("netty:tcp://localhost:{{port}}?sync=false")
+ // continue on another thread so the netty I/O thread
is free to read the reply of the server
+ .threads(1)
+ // the body must stay the request, also after the
reply of the server has arrived
+ .process(exchange -> await().during(1,
TimeUnit.SECONDS).atMost(5, TimeUnit.SECONDS)
+ .until(() -> "Hello
World".equals(exchange.getMessage().getBody(String.class))))
+ .to("mock:result");
+
+ from("netty:tcp://localhost:{{port}}?sync=true")
+ .to("mock:server")
+ .setBody(constant("Bye World"));
+ }
+ };
+ }
+}
diff --git
a/components/camel-netty/src/test/java/org/apache/camel/component/netty/NettyUseByteBufTest.java
b/components/camel-netty/src/test/java/org/apache/camel/component/netty/NettyUseByteBufTest.java
new file mode 100644
index 000000000000..0e1c63fd6fe6
--- /dev/null
+++
b/components/camel-netty/src/test/java/org/apache/camel/component/netty/NettyUseByteBufTest.java
@@ -0,0 +1,86 @@
+/*
+ * 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;
+
+import org.apache.camel.CamelExchangeException;
+import org.apache.camel.CamelExecutionException;
+import org.apache.camel.InvalidPayloadException;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.AvailablePortFinder;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * With useByteBuf=true a body that cannot be converted to a ByteBuf fails the
exchange instead of sending nothing.
+ */
+class NettyUseByteBufTest extends BaseNettyTest {
+
+ @RegisterExtension
+ AvailablePortFinder.Port replyPort = AvailablePortFinder.find();
+
+ @Test
+ void aStringBodyIsSent() throws Exception {
+ getMockEndpoint("mock:result").expectedBodiesReceived("Hello World");
+
+
template.sendBody("netty:tcp://localhost:{{port}}?sync=false&useByteBuf=true",
"Hello World");
+
+ MockEndpoint.assertIsSatisfied(context);
+ }
+
+ @Test
+ void aBodyThatCannotBeConvertedFailsTheExchange() {
+ CamelExecutionException e = assertThrows(CamelExecutionException.class,
+ () ->
template.sendBody("netty:tcp://localhost:{{port}}?sync=false&useByteBuf=true",
new Order()));
+ assertInstanceOf(InvalidPayloadException.class, e.getCause());
+ }
+
+ @Test
+ void aReplyThatCannotBeConvertedClosesTheChannel() {
+ // the client is not left waiting until its request timeout: the
consumer closes the channel
+ CamelExecutionException e = assertThrows(CamelExecutionException.class,
+ () -> template.requestBody("netty:tcp://localhost:" +
replyPort.getPort() + "?sync=true&requestTimeout=5000",
+ "Hello World"));
+ CamelExchangeException cause =
assertInstanceOf(CamelExchangeException.class, e.getCause());
+ assertTrue(cause.getMessage().startsWith("No response received from
remote server"), cause.getMessage());
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("netty:tcp://localhost:{{port}}?sync=false")
+ .to("mock:result");
+
+
fromF("netty:tcp://localhost:%s?sync=true&useByteBuf=true&disconnectOnNoReply=false",
replyPort.getPort())
+ .process(exchange -> exchange.getMessage().setBody(new
Order()));
+ }
+ };
+ }
+
+ private static final class Order {
+ @Override
+ public String toString() {
+ return "Order";
+ }
+ }
+}
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 f70dd6e55b38..a5bf98a4b023 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
@@ -1750,6 +1750,22 @@ exception of the route. With an encoder that accepts any
object, the serialised
message, cause or stack trace of the exception of the route. Set
`muteException=false` to write the exception as before.
`transferExchange=true` is not affected.
+=== camel-netty - replies with the default codec, useByteBuf and sync=false
+
+With the default codec (no `textline`, no custom `encoders` or `decoders`) a
consumer with `sync=true` now converts a
+reply that is not a `String` (or a `ByteBuf`) to a `String` before writing it,
as with `textline=true`. Before, such a
+reply, including the exception of a failed exchange, could not be written and
the client got no reply. A failed exchange
+now replies `java.lang.Exception: Exchange processing failed` (see
`muteException` above). When the reply cannot be
+written, the consumer closes the TCP channel instead of leaving the client
waiting.
+
+With `useByteBuf=true` a body that cannot be converted to a `ByteBuf` (such as
a `File`, a `Map` or a POJO) now fails
+the exchange with an `InvalidPayloadException`, where nothing was sent before.
A consumer with `useByteBuf=true` now
+also converts a reply set as the out message to a `ByteBuf`.
`NettyPayloadHelper.getIn` and `getOut` declare
+`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.
+
=== camel-telegram - webhook requests are verified with a secret token
In webhook mode, Camel now registers the webhook with a secret token, which
Telegram sends in the
