This is an automated email from the ASF dual-hosted git repository.
gnodet pushed a commit to branch camel-4.22.x
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/camel-4.22.x by this push:
new 789d9c8232cf [backport camel-4.22.x] CAMEL-26587: camel-netty - check
transferExchange before processing transferred exchange objects (#29188)
789d9c8232cf is described below
commit 789d9c8232cf6d6e2fbca44ab469f2a99f21107f
Author: Guillaume Nodet <[email protected]>
AuthorDate: Wed Nov 4 21:16:04 2026 +0100
[backport camel-4.22.x] CAMEL-26587: camel-netty - check transferExchange
before processing transferred exchange objects (#29188)
---
.../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 3afd44e3029f..fabd1efa9091 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
