This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch CAMEL-26136-netty-reply-codec in repository https://gitbox.apache.org/repos/asf/camel.git
commit 41d19963af90eb89331ea18f1f6a5d2bacf101d0 Author: Claus Ibsen <[email protected]> AuthorDate: Mon Nov 2 17:13:50 2026 +0100 CAMEL-26136: camel-netty - convert replies for the default codec, fail 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 1bc188419d61..b75d56be558e 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 @@ -1664,6 +1664,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
