This is an automated email from the ASF dual-hosted git repository. gnodet pushed a commit to branch backport/29165-to-camel-4.22.x in repository https://gitbox.apache.org/repos/asf/camel.git
commit 30aba745969fcb72b0e1f934924823abbb4b5d8d Author: Guillaume Nodet <[email protected]> AuthorDate: Wed Nov 4 19:03:40 2026 +0000 [backport camel-4.22.x] CAMEL-26587: camel-netty - check transferExchange before processing transferred exchange objects --- .../camel/component/netty/NettyPayloadHelper.java | 75 +++++++++-- .../netty/handlers/ClientChannelHandler.java | 4 +- .../netty/handlers/ServerChannelHandler.java | 2 +- .../netty/NettyTransferExchangeDisabledTest.java | 149 +++++++++++++++++++++ .../ROOT/pages/camel-4x-upgrade-guide-4_22.adoc | 8 ++ 5 files changed, 225 insertions(+), 13 deletions(-) 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 5dfddf03c45d..c1322fb1e882 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 @@ -63,15 +63,24 @@ public final class NettyPayloadHelper { } } - public static void setIn(Exchange exchange, Object payload) { - if (payload instanceof DefaultExchangeHolder defaultExchangeHolder) { - unmarshal(exchange, defaultExchangeHolder); + /** + * Sets the inbound message of the exchange from a decoded Netty payload. A {@link DefaultExchangeHolder} rebuilds + * the exchange only when the endpoint has {@code transferExchange=true}; otherwise it is set as the message body + * like any other payload. An {@link AddressedEnvelope} (UDP) is unwrapped and its sender address is recorded. + * + * @param endpoint the endpoint that received the payload, whose configuration decides on the exchange transfer + * @param exchange the exchange to set the message on + * @param payload the decoded payload + */ + public static void setIn(NettyEndpoint endpoint, Exchange exchange, Object payload) { + if (isExchangeTransfer(endpoint, payload)) { + unmarshal(exchange, (DefaultExchangeHolder) payload); } else if (payload instanceof AddressedEnvelope) { @SuppressWarnings("unchecked") AddressedEnvelope<Object, InetSocketAddress> dp = (AddressedEnvelope<Object, InetSocketAddress>) payload; // need to check if the content is ExchangeHolder - if (dp.content() instanceof DefaultExchangeHolder defaultExchangeHolder) { - unmarshal(exchange, defaultExchangeHolder); + if (isExchangeTransfer(endpoint, dp.content())) { + unmarshal(exchange, (DefaultExchangeHolder) dp.content()); } else { // need to take out the payload here exchange.getIn().setBody(dp.content()); @@ -86,15 +95,23 @@ public final class NettyPayloadHelper { } } - public static void setOut(Exchange exchange, Object payload) { - if (payload instanceof DefaultExchangeHolder defaultExchangeHolder) { - unmarshal(exchange, defaultExchangeHolder); + /** + * Sets the outbound message of the exchange from a decoded Netty payload, as {@link #setIn} does for the inbound + * message; a payload that is a plain body keeps the headers of the inbound message. + * + * @param endpoint the endpoint that received the payload, whose configuration decides on the exchange transfer + * @param exchange the exchange to set the message on + * @param payload the decoded payload + */ + public static void setOut(NettyEndpoint endpoint, Exchange exchange, Object payload) { + if (isExchangeTransfer(endpoint, payload)) { + unmarshal(exchange, (DefaultExchangeHolder) payload); } else if (payload instanceof AddressedEnvelope) { @SuppressWarnings("unchecked") AddressedEnvelope<Object, InetSocketAddress> dp = (AddressedEnvelope<Object, InetSocketAddress>) payload; // need to check if the content is ExchangeHolder - if (dp.content() instanceof DefaultExchangeHolder defaultExchangeHolder) { - unmarshal(exchange, defaultExchangeHolder); + if (isExchangeTransfer(endpoint, dp.content())) { + unmarshal(exchange, (DefaultExchangeHolder) dp.content()); } else { // need to take out the payload here exchange.getOut().setBody(dp.content()); @@ -108,6 +125,44 @@ public final class NettyPayloadHelper { } } + /** + * A decoded {@link DefaultExchangeHolder} rebuilds the entire Exchange (id, body, headers, out message, properties + * and variables), so it is only unmarshalled when the endpoint has explicitly enabled exchange transfer via + * {@code transferExchange=true}. Otherwise the payload is treated as a normal message body. + */ + private static boolean isExchangeTransfer(NettyEndpoint endpoint, Object payload) { + return payload instanceof DefaultExchangeHolder + && endpoint != null && endpoint.getConfiguration().isTransferExchange(); + } + + /** + * @deprecated use {@link #setIn(NettyEndpoint, Exchange, Object)}. This overload has no endpoint and so + * cannot check {@code transferExchange}: since 4.24.0 it never unmarshals a + * {@link DefaultExchangeHolder} and sets any payload, a holder included, as the message body. + * A caller that relied on the unconditional unmarshalling of earlier versions has to pass its + * endpoint to the three-argument overload. + * @param exchange the exchange + * @param payload the decoded payload + */ + @Deprecated(since = "4.24.0", forRemoval = true) + public static void setIn(Exchange exchange, Object payload) { + setIn(null, exchange, payload); + } + + /** + * @deprecated use {@link #setOut(NettyEndpoint, Exchange, Object)}. This overload has no endpoint and so + * cannot check {@code transferExchange}: since 4.24.0 it never unmarshals a + * {@link DefaultExchangeHolder} and sets any payload, a holder included, as the message body. + * A caller that relied on the unconditional unmarshalling of earlier versions has to pass its + * endpoint to the three-argument overload. + * @param exchange the exchange + * @param payload the decoded payload + */ + @Deprecated(since = "4.24.0", forRemoval = true) + public static void setOut(Exchange exchange, Object payload) { + setOut(null, exchange, payload); + } + /** * The headers and exchange properties that describe the Netty connection of this side. */ 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..d98996cb1551 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 @@ -257,10 +257,10 @@ public class ClientChannelHandler extends SimpleChannelInboundHandler<Object> { // set the result on either IN or OUT on the original exchange depending on its pattern if (ExchangeHelper.isOutCapable(exchange)) { - NettyPayloadHelper.setOut(exchange, body); + NettyPayloadHelper.setOut(producer.getEndpoint(), exchange, body); return exchange.getOut(); } else { - NettyPayloadHelper.setIn(exchange, body); + NettyPayloadHelper.setIn(producer.getEndpoint(), exchange, body); return exchange.getIn(); } } 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 b07d75e0b1ad..256f036abb1c 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 @@ -129,7 +129,7 @@ public class ServerChannelHandler extends SimpleChannelInboundHandler<Object> { // must be prototype scoped (not pooled) so we create the exchange via endpoint Exchange exchange = consumer.createExchange(false); consumer.getEndpoint().updateMessageHeader(exchange.getIn(), ctx); - NettyPayloadHelper.setIn(exchange, message); + NettyPayloadHelper.setIn(consumer.getEndpoint(), exchange, message); return exchange; } diff --git a/components/camel-netty/src/test/java/org/apache/camel/component/netty/NettyTransferExchangeDisabledTest.java b/components/camel-netty/src/test/java/org/apache/camel/component/netty/NettyTransferExchangeDisabledTest.java new file mode 100644 index 000000000000..829fff44be15 --- /dev/null +++ b/components/camel-netty/src/test/java/org/apache/camel/component/netty/NettyTransferExchangeDisabledTest.java @@ -0,0 +1,149 @@ +/* + * 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 io.netty.channel.ChannelHandler; +import io.netty.handler.codec.serialization.ClassResolvers; +import org.apache.camel.BindToRegistry; +import org.apache.camel.Endpoint; +import org.apache.camel.Exchange; +import org.apache.camel.Producer; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.apache.camel.component.netty.codec.ObjectDecoder; +import org.apache.camel.component.netty.codec.ObjectEncoder; +import org.apache.camel.support.DefaultExchange; +import org.apache.camel.support.DefaultExchangeHolder; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class NettyTransferExchangeDisabledTest extends BaseNettyTest { + + @BindToRegistry("encoder") + public ChannelHandler getEncoder() { + return new ShareableChannelHandlerFactory(new ObjectEncoder()); + } + + @BindToRegistry("decoder") + public ChannelHandler getDecoder() { + return new DefaultChannelHandlerFactory() { + @Override + public ChannelHandler newChannelHandler() { + return new ObjectDecoder(ClassResolvers.weakCachingResolver(null)); + } + }; + } + + @Test + public void testTransferExchangeDisabledTreatsHolderAsBody() throws Exception { + MockEndpoint mock = getMockEndpoint("mock:result"); + mock.expectedMessageCount(1); + + Exchange sourceExchange = new DefaultExchange(context); + sourceExchange.getIn().setBody("Inner Payload"); + sourceExchange.getIn().setHeader("CustomHeader", "HeaderValue"); + sourceExchange.setProperty("CustomProperty", "PropertyValue"); + + DefaultExchangeHolder holder = DefaultExchangeHolder.marshal(sourceExchange, true, true); + + Endpoint endpoint + = context.getEndpoint("netty:tcp://localhost:{{port}}?sync=false&encoders=#encoder&decoders=#decoder"); + Producer producer = endpoint.createProducer(); + producer.start(); + try { + Exchange sendExchange = endpoint.createExchange(); + sendExchange.getIn().setBody(holder); + producer.process(sendExchange); + } finally { + producer.stop(); + } + + mock.assertIsSatisfied(); + + Exchange received = mock.getReceivedExchanges().get(0); + assertNotNull(received); + + // Holder must remain the body and NOT be unmarshalled into the exchange + assertTrue(received.getIn().getBody() instanceof DefaultExchangeHolder, + "Body should remain a DefaultExchangeHolder when transferExchange=false"); + assertNull(received.getIn().getHeader("CustomHeader"), + "Headers from DefaultExchangeHolder must not be unmarshalled when transferExchange=false"); + assertNull(received.getProperty("CustomProperty"), + "Properties from DefaultExchangeHolder must not be unmarshalled when transferExchange=false"); + assertNotNull(received.getIn().getHeader(NettyConstants.NETTY_REMOTE_ADDRESS), + "Connection headers must be preserved"); + } + + @Test + public void testHelperGating() { + NettyEndpoint transferOn = context.getEndpoint("netty:tcp://localhost:1111?transferExchange=true", NettyEndpoint.class); + NettyEndpoint transferOff + = context.getEndpoint("netty:tcp://localhost:2222?transferExchange=false", NettyEndpoint.class); + + Exchange source = new DefaultExchange(context); + source.getIn().setBody("Inner Content"); + source.getIn().setHeader("Key", "Value"); + source.setProperty("Prop", "Val"); + DefaultExchangeHolder holder = DefaultExchangeHolder.marshal(source, true, true); + + // 1. With transferExchange = true + Exchange targetOn = new DefaultExchange(context); + NettyPayloadHelper.setIn(transferOn, targetOn, holder); + assertEquals("Inner Content", targetOn.getIn().getBody()); + assertEquals("Value", targetOn.getIn().getHeader("Key")); + assertEquals("Val", targetOn.getProperty("Prop")); + + // 2. With transferExchange = false + Exchange targetOff = new DefaultExchange(context); + NettyPayloadHelper.setIn(transferOff, targetOff, holder); + assertSame(holder, targetOff.getIn().getBody()); + assertNull(targetOff.getIn().getHeader("Key")); + assertNull(targetOff.getProperty("Prop")); + + // 3. With null endpoint (deprecated overload) + Exchange targetNull = new DefaultExchange(context); + NettyPayloadHelper.setIn(targetNull, holder); + assertSame(holder, targetNull.getIn().getBody()); + assertNull(targetNull.getIn().getHeader("Key")); + assertNull(targetNull.getProperty("Prop")); + + // 4. setOut with transferExchange = false + Exchange targetOutOff = new DefaultExchange(context); + targetOutOff.getIn().setHeader("ExistingHeader", "ExistingVal"); + NettyPayloadHelper.setOut(transferOff, targetOutOff, holder); + assertSame(holder, targetOutOff.getMessage().getBody()); + assertNull(targetOutOff.getMessage().getHeader("Key")); + assertEquals("ExistingVal", targetOutOff.getMessage().getHeader("ExistingHeader")); + assertNull(targetOutOff.getProperty("Prop")); + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + from("netty:tcp://localhost:{{port}}?sync=false&encoders=#encoder&decoders=#decoder") + .to("mock:result"); + } + }; + } +} diff --git a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc index 4ffee56c75ca..21b0b8f23289 100644 --- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc +++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc @@ -197,6 +197,14 @@ a header in the transferred exchange is no longer used, also when the receiving the `CamelNettySSLClientCert*` headers on a connection without a client certificate). This also applies to the reply the producer receives. Send any such value of the sender under a header name of your own. +=== camel-netty - transferExchange requires explicit endpoint configuration + +`NettyPayloadHelper` now checks `transferExchange=true` on the endpoint configuration before reconstructing exchange data from a `DefaultExchangeHolder` payload. + +Before, a receiver with an object decoder silently reconstructed the exchange even when `transferExchange=false`. Now, when `transferExchange` is disabled (the default), the incoming `DefaultExchangeHolder` is left as an opaque body (both on the consumer side and for the producer's reply via `setOut`). + +In addition, the legacy two-argument methods `NettyPayloadHelper.setIn(Exchange, Object)` and `setOut(Exchange, Object)` are deprecated and no longer unmarshal `DefaultExchangeHolder` payloads. Callers should migrate to the three-argument overloads passing the `NettyEndpoint`. + === camel-flowable - xml inbound channels parse the event with the Camel XML settings The `flowable` producer now parses the event of an inbound channel of type `camel` with `deserializerType` set to
