This is an automated email from the ASF dual-hosted git repository.
davsclaus 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 3bb288569a0f [backport camel-4.22.x] CAMEL-26278: camel-hazelcast -
the hazelcast-seda consumer restores a transferred exchange only with
transferExchange=true and otherwise takes its body (#29614)
3bb288569a0f is described below
commit 3bb288569a0f9c0012a0fa573429bf3ca3129af1
Author: Andrea Cosentino <[email protected]>
AuthorDate: Wed Nov 11 10:28:51 2026 +0100
[backport camel-4.22.x] CAMEL-26278: camel-hazelcast - the hazelcast-seda
consumer restores a transferred exchange only with transferExchange=true and
otherwise takes its body (#29614)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
Signed-off-by: Andrea Cosentino <[email protected]>
---
.../camel/catalog/components/hazelcast-seda.json | 2 +-
.../catalog/docs/hazelcast-seda-component.adoc | 27 ++++++++
.../component/hazelcast/seda/hazelcast-seda.json | 2 +-
.../src/main/docs/hazelcast-seda-component.adoc | 27 ++++++++
.../hazelcast/seda/HazelcastSedaConfiguration.java | 3 +-
.../hazelcast/seda/HazelcastSedaConsumer.java | 26 ++++++-
.../HazelcastSedaTransferExchangeTest.java | 80 ++++++++++++++++++++--
.../dsl/HazelcastSedaEndpointBuilderFactory.java | 30 ++++++--
8 files changed, 181 insertions(+), 16 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/hazelcast-seda.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/hazelcast-seda.json
index 295869b94e03..c0bfbde7e259 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/hazelcast-seda.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/hazelcast-seda.json
@@ -44,6 +44,6 @@
"onErrorDelay": { "index": 10, "kind": "parameter", "displayName": "On
Error Delay", "group": "seda", "label": "seda", "required": false, "type":
"integer", "javaType": "int", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 1000, "configurationClass":
"org.apache.camel.component.hazelcast.seda.HazelcastSedaConfiguration",
"configurationField": "configuration", "description": "Milliseconds before
consumer continues polling after an error has occurred." },
"pollTimeout": { "index": 11, "kind": "parameter", "displayName": "Poll
Timeout", "group": "seda", "label": "seda", "required": false, "type":
"integer", "javaType": "int", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 1000, "configurationClass":
"org.apache.camel.component.hazelcast.seda.HazelcastSedaConfiguration",
"configurationField": "configuration", "description": "The timeout used when
consuming from the SEDA queue. When a timeout occurs, the consum [...]
"transacted": { "index": 12, "kind": "parameter", "displayName":
"Transacted", "group": "seda", "label": "seda", "required": false, "type":
"boolean", "javaType": "boolean", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": false, "configurationClass":
"org.apache.camel.component.hazelcast.seda.HazelcastSedaConfiguration",
"configurationField": "configuration", "description": "If set to true then the
consumer runs in transaction mode, where the messages in the [...]
- "transferExchange": { "index": 13, "kind": "parameter", "displayName":
"Transfer Exchange", "group": "seda", "label": "seda", "required": false,
"type": "boolean", "javaType": "boolean", "deprecated": false, "autowired":
false, "secret": false, "security": "insecure:serialization", "defaultValue":
false, "configurationClass":
"org.apache.camel.component.hazelcast.seda.HazelcastSedaConfiguration",
"configurationField": "configuration", "description": "If set to true the whole
Exchange [...]
+ "transferExchange": { "index": 13, "kind": "parameter", "displayName":
"Transfer Exchange", "group": "seda", "label": "seda", "required": false,
"type": "boolean", "javaType": "boolean", "deprecated": false, "autowired":
false, "secret": false, "security": "insecure:serialization", "defaultValue":
false, "configurationClass":
"org.apache.camel.component.hazelcast.seda.HazelcastSedaConfiguration",
"configurationField": "configuration", "description": "If set to true the whole
Exchange [...]
}
}
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/hazelcast-seda-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/hazelcast-seda-component.adoc
index 550d5c0c922f..a84eb7946866 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/hazelcast-seda-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/hazelcast-seda-component.adoc
@@ -101,3 +101,30 @@ YAML::
----
====
+== Transferring the whole exchange
+
+By default only the message body goes on the queue. With
`transferExchange=true` the producer puts the whole exchange
+on the queue: the body, and the headers and exchange properties whose values
are text, numbers, booleans or dates. A
+consumer with `transferExchange=true` restores them. A consumer without the
option takes only the body of such an
+exchange: set `transferExchange=true` on both endpoints to keep the headers
and exchange properties.
+
+[source,yaml]
+----
+- route:
+ from:
+ uri: direct:foo
+ steps:
+ - to:
+ uri: hazelcast-seda:foo
+ parameters:
+ transferExchange: true
+- route:
+ from:
+ uri: hazelcast-seda:foo
+ parameters:
+ transferExchange: true
+ steps:
+ - to:
+ uri: mock:result
+----
+
diff --git
a/components/camel-hazelcast/src/generated/resources/META-INF/org/apache/camel/component/hazelcast/seda/hazelcast-seda.json
b/components/camel-hazelcast/src/generated/resources/META-INF/org/apache/camel/component/hazelcast/seda/hazelcast-seda.json
index 295869b94e03..c0bfbde7e259 100644
---
a/components/camel-hazelcast/src/generated/resources/META-INF/org/apache/camel/component/hazelcast/seda/hazelcast-seda.json
+++
b/components/camel-hazelcast/src/generated/resources/META-INF/org/apache/camel/component/hazelcast/seda/hazelcast-seda.json
@@ -44,6 +44,6 @@
"onErrorDelay": { "index": 10, "kind": "parameter", "displayName": "On
Error Delay", "group": "seda", "label": "seda", "required": false, "type":
"integer", "javaType": "int", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 1000, "configurationClass":
"org.apache.camel.component.hazelcast.seda.HazelcastSedaConfiguration",
"configurationField": "configuration", "description": "Milliseconds before
consumer continues polling after an error has occurred." },
"pollTimeout": { "index": 11, "kind": "parameter", "displayName": "Poll
Timeout", "group": "seda", "label": "seda", "required": false, "type":
"integer", "javaType": "int", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 1000, "configurationClass":
"org.apache.camel.component.hazelcast.seda.HazelcastSedaConfiguration",
"configurationField": "configuration", "description": "The timeout used when
consuming from the SEDA queue. When a timeout occurs, the consum [...]
"transacted": { "index": 12, "kind": "parameter", "displayName":
"Transacted", "group": "seda", "label": "seda", "required": false, "type":
"boolean", "javaType": "boolean", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": false, "configurationClass":
"org.apache.camel.component.hazelcast.seda.HazelcastSedaConfiguration",
"configurationField": "configuration", "description": "If set to true then the
consumer runs in transaction mode, where the messages in the [...]
- "transferExchange": { "index": 13, "kind": "parameter", "displayName":
"Transfer Exchange", "group": "seda", "label": "seda", "required": false,
"type": "boolean", "javaType": "boolean", "deprecated": false, "autowired":
false, "secret": false, "security": "insecure:serialization", "defaultValue":
false, "configurationClass":
"org.apache.camel.component.hazelcast.seda.HazelcastSedaConfiguration",
"configurationField": "configuration", "description": "If set to true the whole
Exchange [...]
+ "transferExchange": { "index": 13, "kind": "parameter", "displayName":
"Transfer Exchange", "group": "seda", "label": "seda", "required": false,
"type": "boolean", "javaType": "boolean", "deprecated": false, "autowired":
false, "secret": false, "security": "insecure:serialization", "defaultValue":
false, "configurationClass":
"org.apache.camel.component.hazelcast.seda.HazelcastSedaConfiguration",
"configurationField": "configuration", "description": "If set to true the whole
Exchange [...]
}
}
diff --git
a/components/camel-hazelcast/src/main/docs/hazelcast-seda-component.adoc
b/components/camel-hazelcast/src/main/docs/hazelcast-seda-component.adoc
index 550d5c0c922f..a84eb7946866 100644
--- a/components/camel-hazelcast/src/main/docs/hazelcast-seda-component.adoc
+++ b/components/camel-hazelcast/src/main/docs/hazelcast-seda-component.adoc
@@ -101,3 +101,30 @@ YAML::
----
====
+== Transferring the whole exchange
+
+By default only the message body goes on the queue. With
`transferExchange=true` the producer puts the whole exchange
+on the queue: the body, and the headers and exchange properties whose values
are text, numbers, booleans or dates. A
+consumer with `transferExchange=true` restores them. A consumer without the
option takes only the body of such an
+exchange: set `transferExchange=true` on both endpoints to keep the headers
and exchange properties.
+
+[source,yaml]
+----
+- route:
+ from:
+ uri: direct:foo
+ steps:
+ - to:
+ uri: hazelcast-seda:foo
+ parameters:
+ transferExchange: true
+- route:
+ from:
+ uri: hazelcast-seda:foo
+ parameters:
+ transferExchange: true
+ steps:
+ - to:
+ uri: mock:result
+----
+
diff --git
a/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/seda/HazelcastSedaConfiguration.java
b/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/seda/HazelcastSedaConfiguration.java
index 4ad12767cc17..c01eada714b9 100644
---
a/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/seda/HazelcastSedaConfiguration.java
+++
b/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/seda/HazelcastSedaConfiguration.java
@@ -109,7 +109,8 @@ public class HazelcastSedaConfiguration {
/**
* If set to true the whole Exchange will be transfered. If header or body
contains not serializable objects, they
- * will be skipped.
+ * will be skipped. Set it on both the producer and the consumer: the
producer puts the whole Exchange on the queue,
+ * and the consumer restores its headers and exchange properties. A
consumer without it takes only the body.
*/
public void setTransferExchange(boolean transferExchange) {
this.transferExchange = transferExchange;
diff --git
a/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/seda/HazelcastSedaConsumer.java
b/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/seda/HazelcastSedaConsumer.java
index bae894aebd1e..e78fdc8e01e3 100644
---
a/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/seda/HazelcastSedaConsumer.java
+++
b/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/seda/HazelcastSedaConsumer.java
@@ -31,6 +31,7 @@ import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.support.AsyncProcessorConverterHelper;
import org.apache.camel.support.DefaultConsumer;
+import org.apache.camel.support.DefaultExchange;
import org.apache.camel.support.DefaultExchangeHolder;
import org.apache.camel.support.task.Tasks;
import org.apache.camel.support.task.budget.Budgets;
@@ -48,6 +49,7 @@ public class HazelcastSedaConsumer extends DefaultConsumer
implements Runnable {
private final AsyncProcessor processor;
private ExecutorService executor;
private ScheduledExecutorService taskExecutor;
+ private volatile boolean transferredExchangeLogged;
public HazelcastSedaConsumer(final Endpoint endpoint, final Processor
processor) {
super(endpoint, processor);
@@ -105,8 +107,8 @@ public class HazelcastSedaConsumer extends DefaultConsumer
implements Runnable {
final Object body =
queue.poll(endpoint.getConfiguration().getPollTimeout(), TimeUnit.MILLISECONDS);
if (body != null) {
- if (body instanceof DefaultExchangeHolder
defaultExchangeHolder) {
- DefaultExchangeHolder.unmarshal(exchange,
defaultExchangeHolder);
+ if (body instanceof DefaultExchangeHolder holder) {
+ takeTransferredExchange(exchange, holder);
} else {
exchange.getIn().setBody(body);
}
@@ -176,4 +178,24 @@ public class HazelcastSedaConsumer extends DefaultConsumer
implements Runnable {
}
}
+ /**
+ * An exchange that a producer with transferExchange=true put on the
queue: the consumer restores its headers and
+ * exchange properties only when it has transferExchange=true as well, and
otherwise takes only its body.
+ */
+ private void takeTransferredExchange(Exchange exchange,
DefaultExchangeHolder holder) {
+ if (endpoint.getConfiguration().isTransferExchange()) {
+ DefaultExchangeHolder.unmarshal(exchange, holder);
+ return;
+ }
+ if (!transferredExchangeLogged) {
+ transferredExchangeLogged = true;
+ LOG.warn("The queue {} holds an exchange that a producer with
transferExchange=true put on it, but this consumer"
+ + " has transferExchange=false: it takes only the body of
such an exchange, without its headers and"
+ + " exchange properties. Set transferExchange=true on the
consumer to restore them.",
+ endpoint.getConfiguration().getQueueName());
+ }
+ Exchange transferred = new DefaultExchange(endpoint.getCamelContext());
+ DefaultExchangeHolder.unmarshal(transferred, holder);
+ exchange.getIn().setBody(transferred.getIn().getBody());
+ }
}
diff --git
a/components/camel-hazelcast/src/test/java/org/apache/camel/component/hazelcast/HazelcastSedaTransferExchangeTest.java
b/components/camel-hazelcast/src/test/java/org/apache/camel/component/hazelcast/HazelcastSedaTransferExchangeTest.java
index 39b57e5c73bf..ca7d3659b41f 100644
---
a/components/camel-hazelcast/src/test/java/org/apache/camel/component/hazelcast/HazelcastSedaTransferExchangeTest.java
+++
b/components/camel-hazelcast/src/test/java/org/apache/camel/component/hazelcast/HazelcastSedaTransferExchangeTest.java
@@ -23,6 +23,7 @@ import org.apache.camel.EndpointInject;
import org.apache.camel.Exchange;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.support.DefaultExchangeHolder;
import org.apache.camel.test.infra.common.TestEntityNameGenerator;
import org.apache.camel.test.infra.hazelcast.services.HazelcastService;
import org.apache.camel.test.infra.hazelcast.services.HazelcastServiceFactory;
@@ -34,6 +35,8 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestInstance;
import org.junit.jupiter.api.extension.RegisterExtension;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
public class HazelcastSedaTransferExchangeTest extends CamelTestSupport {
@@ -46,6 +49,12 @@ public class HazelcastSedaTransferExchangeTest extends
CamelTestSupport {
@EndpointInject("mock:result")
private MockEndpoint mock;
+ @EndpointInject("mock:transferred")
+ private MockEndpoint transferred;
+
+ @EndpointInject("mock:bodyOnly")
+ private MockEndpoint bodyOnly;
+
private HazelcastInstance hazelcastInstance;
@BeforeAll
@@ -56,6 +65,8 @@ public class HazelcastSedaTransferExchangeTest extends
CamelTestSupport {
@AfterEach
public void afterEach() {
mock.reset();
+ transferred.reset();
+ bodyOnly.reset();
}
@AfterAll
@@ -69,18 +80,69 @@ public class HazelcastSedaTransferExchangeTest extends
CamelTestSupport {
public void testExchangeTransferEnabled() throws InterruptedException {
final String value = "CAMEL-3983";
- mock.expectedMessageCount(1);
- mock.expectedBodiesReceived("test");
- mock.expectedHeaderReceived("test", value);
+ transferred.expectedMessageCount(1);
+ transferred.expectedBodiesReceived("test");
+ transferred.expectedHeaderReceived("test", value);
+ transferred.expectedPropertyReceived("test", value);
Exchange exchange = createExchangeWithBody("test");
exchange.getIn().setHeader("test", value);
+ exchange.setProperty("test", value);
template.send("direct:foobar", exchange);
MockEndpoint.assertIsSatisfied(context);
}
+ @Test
+ void
transferredExchangeGivesOnlyItsBodyToAConsumerWithoutTransferExchange() throws
InterruptedException {
+ bodyOnly.expectedMessageCount(1);
+ bodyOnly.expectedBodiesReceived("test");
+ bodyOnly.expectedNoHeaderReceived();
+
+ Exchange exchange = createExchangeWithBody("test");
+ exchange.getIn().setHeader("test", "Not propagated");
+ exchange.setProperty("test", "Not propagated");
+
+ template.send("direct:bodyOnly", exchange);
+
+ MockEndpoint.assertIsSatisfied(context);
+ assertNull(bodyOnly.getReceivedExchanges().get(0).getProperty("test"));
+ }
+
+ @Test
+ void holderOnTheQueueGivesOnlyItsBodyWithoutTransferExchange() throws
InterruptedException {
+ bodyOnly.expectedMessageCount(1);
+ bodyOnly.expectedBodiesReceived("held");
+ bodyOnly.expectedNoHeaderReceived();
+
+ hazelcastInstance.getQueue("body-only").add(holder("held"));
+
+ MockEndpoint.assertIsSatisfied(context);
+ Exchange received = bodyOnly.getReceivedExchanges().get(0);
+ assertNull(received.getProperty("test"));
+ assertNull(received.getException());
+ }
+
+ @Test
+ void holderOnTheQueueGivesItsHeadersAndPropertiesWithTransferExchange()
throws InterruptedException {
+ transferred.expectedMessageCount(1);
+ transferred.expectedBodiesReceived("held");
+ transferred.expectedHeaderReceived("test", "held header");
+ transferred.expectedPropertyReceived("test", "held property");
+
+ hazelcastInstance.getQueue("transferred").add(holder("held"));
+
+ MockEndpoint.assertIsSatisfied(context);
+ }
+
+ private DefaultExchangeHolder holder(String body) {
+ Exchange exchange = createExchangeWithBody(body);
+ exchange.getIn().setHeader("test", "held header");
+ exchange.setProperty("test", "held property");
+ return DefaultExchangeHolder.marshal(exchange);
+ }
+
@Test
public void testExchangeTransferDisabled() throws InterruptedException {
mock.expectedMessageCount(1);
@@ -110,9 +172,17 @@ public class HazelcastSedaTransferExchangeTest extends
CamelTestSupport {
public void configure() {
from("direct:foo").to("hazelcast-seda:foo");
-
from("direct:foobar").to("hazelcast-seda:foo?transferExchange=true");
-
from("hazelcast-seda:foo").to("mock:result");
+
+ // transferExchange on both sides: the headers and exchange
properties come along
+
from("direct:foobar").to("hazelcast-seda:transferred?transferExchange=true");
+
+
from("hazelcast-seda:transferred?transferExchange=true").to("mock:transferred");
+
+ // transferExchange only on the producer: the consumer takes
only the body
+
from("direct:bodyOnly").to("hazelcast-seda:body-only?transferExchange=true");
+
+ from("hazelcast-seda:body-only").to("mock:bodyOnly");
}
};
}
diff --git
a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/HazelcastSedaEndpointBuilderFactory.java
b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/HazelcastSedaEndpointBuilderFactory.java
index dad4ad4a5003..3dbeda3d7718 100644
---
a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/HazelcastSedaEndpointBuilderFactory.java
+++
b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/HazelcastSedaEndpointBuilderFactory.java
@@ -277,7 +277,10 @@ public interface HazelcastSedaEndpointBuilderFactory {
}
/**
* If set to true the whole Exchange will be transfered. If header or
- * body contains not serializable objects, they will be skipped.
+ * body contains not serializable objects, they will be skipped. Set it
+ * on both the producer and the consumer: the producer puts the whole
+ * Exchange on the queue, and the consumer restores its headers and
+ * exchange properties. A consumer without it takes only the body.
*
* The option is a: <code>boolean</code> type.
*
@@ -293,7 +296,10 @@ public interface HazelcastSedaEndpointBuilderFactory {
}
/**
* If set to true the whole Exchange will be transfered. If header or
- * body contains not serializable objects, they will be skipped.
+ * body contains not serializable objects, they will be skipped. Set it
+ * on both the producer and the consumer: the producer puts the whole
+ * Exchange on the queue, and the consumer restores its headers and
+ * exchange properties. A consumer without it takes only the body.
*
* The option will be converted to a <code>boolean</code> type.
*
@@ -680,7 +686,10 @@ public interface HazelcastSedaEndpointBuilderFactory {
}
/**
* If set to true the whole Exchange will be transfered. If header or
- * body contains not serializable objects, they will be skipped.
+ * body contains not serializable objects, they will be skipped. Set it
+ * on both the producer and the consumer: the producer puts the whole
+ * Exchange on the queue, and the consumer restores its headers and
+ * exchange properties. A consumer without it takes only the body.
*
* The option is a: <code>boolean</code> type.
*
@@ -696,7 +705,10 @@ public interface HazelcastSedaEndpointBuilderFactory {
}
/**
* If set to true the whole Exchange will be transfered. If header or
- * body contains not serializable objects, they will be skipped.
+ * body contains not serializable objects, they will be skipped. Set it
+ * on both the producer and the consumer: the producer puts the whole
+ * Exchange on the queue, and the consumer restores its headers and
+ * exchange properties. A consumer without it takes only the body.
*
* The option will be converted to a <code>boolean</code> type.
*
@@ -1012,7 +1024,10 @@ public interface HazelcastSedaEndpointBuilderFactory {
}
/**
* If set to true the whole Exchange will be transfered. If header or
- * body contains not serializable objects, they will be skipped.
+ * body contains not serializable objects, they will be skipped. Set it
+ * on both the producer and the consumer: the producer puts the whole
+ * Exchange on the queue, and the consumer restores its headers and
+ * exchange properties. A consumer without it takes only the body.
*
* The option is a: <code>boolean</code> type.
*
@@ -1028,7 +1043,10 @@ public interface HazelcastSedaEndpointBuilderFactory {
}
/**
* If set to true the whole Exchange will be transfered. If header or
- * body contains not serializable objects, they will be skipped.
+ * body contains not serializable objects, they will be skipped. Set it
+ * on both the producer and the consumer: the producer puts the whole
+ * Exchange on the queue, and the consumer restores its headers and
+ * exchange properties. A consumer without it takes only the body.
*
* The option will be converted to a <code>boolean</code> type.
*
