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 d4688eb0bb38 CAMEL-25816: camel-hazelcast - the transacted SEDA
consumer ends its transaction on its own thread (#28205)
d4688eb0bb38 is described below
commit d4688eb0bb38a776f428b464d86c2b613fd4d29c
Author: allthingssecurity <[email protected]>
AuthorDate: Thu Oct 22 00:36:27 2026 +0530
CAMEL-25816: camel-hazelcast - the transacted SEDA consumer ends its
transaction on its own thread (#28205)
* CAMEL-25816: camel-hazelcast - the transacted SEDA consumer ends its
transaction on its own thread
A Hazelcast transaction can only be committed or rolled back by the thread
that began it. When the route of a transacted hazelcast-seda consumer
continued on another thread (an asynchronous producer such as camel-kafka,
or threads()), the done callback tried to commit or roll back there, which
failed with "Transaction cannot span multiple threads!" (a WARN that the
reactive executor ignores). The message stayed reserved by a transaction
that never ended: a failed message was not taken again, and when the
application left the cluster the other members rolled back its open
transactions, so every message processed since the start was put back
into the queue and consumed again.
With transacted=true the consumer thread now waits for the exchange to
complete and commits or rolls back itself. Upgrade note in the 4.24 guide.
* CAMEL-25816: camel-hazelcast - address review: release the transacted
exchange after the commit or rollback
With the pooled exchange factory an auto released exchange is reset when
its unit of work is done, which clears its exception before the
transacted consumer decides to commit or roll back: a failed exchange was
committed and its message lost. The transacted consumer now creates the
exchange without auto release and releases it after the commit or
rollback, as camel-jpa and camel-sql do. The transacted path is chosen
on the transacted option, so an exchange it releases is never routed
asynchronously.
* CAMEL-25816: camel-hazelcast - address review: document the transaction
timeout, stop the context before the members in the tests
The consumer begins its transaction with newTransactionContext(), which
uses TransactionOptions.getDefault() (a 2 minute timeout), and
TransactionImpl.commit() checks the timeout first. Now that a route that
continues on another thread is waited for, an exchange can outlast it:
the commit then fails with TransactionTimedOutException, the
transaction is rolled back and the message is consumed again. The
transacted option description (with the generated JSON and endpoint
DSL) and the upgrade guide note say so.
The tests stop the Camel context before they shut the Hazelcast members
down (JUnit runs the @AfterEach of the test before the extension of
CamelTestSupport stops the context), so the consumers no longer poll
members that are gone, and the new test classes and methods are
package-private.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../camel/catalog/components/hazelcast-seda.json | 2 +-
.../component/hazelcast/seda/hazelcast-seda.json | 2 +-
.../hazelcast/seda/HazelcastSedaConfiguration.java | 6 +-
.../hazelcast/seda/HazelcastSedaConsumer.java | 66 ++++++----
.../HazelcastSedaTransactedAsyncRouteTest.java | 136 +++++++++++++++++++++
.../HazelcastSedaTransactedPooledExchangeTest.java | 34 ++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 14 +++
.../dsl/HazelcastSedaEndpointBuilderFactory.java | 48 +++++++-
8 files changed, 277 insertions(+), 31 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 fc879033caaa..430f9cce531e 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
@@ -43,7 +43,7 @@
"concurrentConsumers": { "index": 9, "kind": "parameter", "displayName":
"Concurrent Consumers", "group": "seda", "label": "seda", "required": false,
"type": "integer", "javaType": "int", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 1, "configurationClass":
"org.apache.camel.component.hazelcast.seda.HazelcastSedaConfiguration",
"configurationField": "configuration", "description": "To use concurrent
consumers polling from the SEDA queue." },
"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 [...]
+ "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 [...]
}
}
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 fc879033caaa..430f9cce531e 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
@@ -43,7 +43,7 @@
"concurrentConsumers": { "index": 9, "kind": "parameter", "displayName":
"Concurrent Consumers", "group": "seda", "label": "seda", "required": false,
"type": "integer", "javaType": "int", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 1, "configurationClass":
"org.apache.camel.component.hazelcast.seda.HazelcastSedaConfiguration",
"configurationField": "configuration", "description": "To use concurrent
consumers polling from the SEDA queue." },
"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 [...]
+ "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 [...]
}
}
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..9c5be6dcd05d 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
@@ -121,7 +121,11 @@ public class HazelcastSedaConfiguration {
/**
* If set to true then the consumer runs in transaction mode, where the
messages in the seda queue will only be
- * removed if the transaction commits, which happens when the processing
is complete.
+ * removed if the transaction commits, which happens when the processing
is complete. A Hazelcast transaction
+ * belongs to the thread that began it, so each consumer thread waits for
its exchange to complete, also when the
+ * route continues on another thread, before it takes the next message.
The exchange must complete within the
+ * Hazelcast transaction timeout (2 minutes by default): when it takes
longer, the commit fails, the transaction is
+ * rolled back and the message is consumed again.
*/
public void setTransacted(boolean transacted) {
this.transacted = transacted;
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 a444a6c19d42..361ae8759d04 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
@@ -87,11 +87,14 @@ public class HazelcastSedaConsumer extends DefaultConsumer
implements Runnable {
BaseQueue<?> queue =
endpoint.getHazelcastInstance().getQueue(endpoint.getConfiguration().getQueueName());
while (queue != null && isRunAllowed()) {
- final Exchange exchange = createExchange(true);
+ final boolean transacted =
endpoint.getConfiguration().isTransacted();
+ // the transacted consumer reads the outcome of the exchange after
the routing is done, so it releases the
+ // exchange itself: a pooled exchange is otherwise reset (and its
exception cleared) when it is done
+ final Exchange exchange = createExchange(!transacted);
TransactionContext transactionCtx = null;
try {
- if (endpoint.getConfiguration().isTransacted()) {
+ if (transacted) {
// Get and begin transaction if exist
transactionCtx =
endpoint.getHazelcastInstance().newTransactionContext();
@@ -112,27 +115,24 @@ public class HazelcastSedaConsumer extends
DefaultConsumer implements Runnable {
}
try {
final TransactionContext txc = transactionCtx;
- // process using the asynchronous routing engine
- processor.process(exchange, new AsyncCallback() {
- public void done(boolean asyncDone) {
- // a rollback only exchange (without an
exception) did not complete successfully either
- boolean rollback = exchange.getException() !=
null || exchange.isRollbackOnly()
- || exchange.isRollbackOnlyLast();
- if (rollback && txc != null) {
- LOG.trace("Rollback transaction: {}",
txc.getTxnId());
- txc.rollbackTransaction();
- }
- if (exchange.getException() != null) {
-
getExceptionHandler().handleException("Error processing exchange", exchange,
- exchange.getException());
- }
- // It's OK, I commit
- if (!rollback && txc != null) {
- LOG.trace("Commit transaction: {}",
txc.getTxnId());
- txc.commitTransaction();
- }
+ if (transacted) {
+ // a Hazelcast transaction belongs to the thread
that began it, and cannot be committed
+ // or rolled back from another thread: wait for
the exchange to complete, also when the
+ // route continues on another thread, and end the
transaction on this thread
+ try {
+ processor.process(exchange);
+ } catch (Exception e) {
+ exchange.setException(e);
}
- });
+ onExchangeDone(exchange, txc);
+ } else {
+ // process using the asynchronous routing engine
+ processor.process(exchange, new AsyncCallback() {
+ public void done(boolean asyncDone) {
+ onExchangeDone(exchange, null);
+ }
+ });
+ }
} catch (Exception e) {
LOG.error("Hzlq Exception caught: {}", e, e);
// Rollback
@@ -175,8 +175,30 @@ public class HazelcastSedaConsumer extends DefaultConsumer
implements Runnable {
.withName("HazelcastSedaErrorRecoveryDelay")
.build()
.run(getEndpoint().getCamelContext(), () -> true);
+ } finally {
+ if (transacted) {
+ releaseExchange(exchange, false);
+ }
}
}
}
+ private void onExchangeDone(Exchange exchange, TransactionContext txc) {
+ // a rollback only exchange (without an exception) did not complete
successfully either
+ boolean rollback = exchange.getException() != null ||
exchange.isRollbackOnly()
+ || exchange.isRollbackOnlyLast();
+ if (rollback && txc != null) {
+ LOG.trace("Rollback transaction: {}", txc.getTxnId());
+ txc.rollbackTransaction();
+ }
+ if (exchange.getException() != null) {
+ getExceptionHandler().handleException("Error processing exchange",
exchange, exchange.getException());
+ }
+ // It's OK, I commit
+ if (!rollback && txc != null) {
+ LOG.trace("Commit transaction: {}", txc.getTxnId());
+ txc.commitTransaction();
+ }
+ }
+
}
diff --git
a/components/camel-hazelcast/src/test/java/org/apache/camel/component/hazelcast/HazelcastSedaTransactedAsyncRouteTest.java
b/components/camel-hazelcast/src/test/java/org/apache/camel/component/hazelcast/HazelcastSedaTransactedAsyncRouteTest.java
new file mode 100644
index 000000000000..5ba93dfaae51
--- /dev/null
+++
b/components/camel-hazelcast/src/test/java/org/apache/camel/component/hazelcast/HazelcastSedaTransactedAsyncRouteTest.java
@@ -0,0 +1,136 @@
+/*
+ * 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.hazelcast;
+
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import com.hazelcast.config.Config;
+import com.hazelcast.core.Hazelcast;
+import com.hazelcast.core.HazelcastInstance;
+import org.apache.camel.CamelContext;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * A transacted Hazelcast SEDA consumer whose route continues on another
thread (here {@code threads()}, as with any
+ * asynchronous producer). A Hazelcast transaction can only be committed or
rolled back by the thread that began it, so
+ * the consumer must end the transaction on its own thread once the exchange
is complete.
+ */
+class HazelcastSedaTransactedAsyncRouteTest extends CamelTestSupport {
+
+ private final AtomicInteger attempts = new AtomicInteger();
+ private HazelcastInstance member;
+ private HazelcastInstance otherMember;
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ String clusterName = "seda-transacted-async-" + UUID.randomUUID();
+ otherMember =
Hazelcast.newHazelcastInstance(memberConfig(clusterName));
+ member = Hazelcast.newHazelcastInstance(memberConfig(clusterName));
+ await().atMost(30, TimeUnit.SECONDS).until(() ->
member.getCluster().getMembers().size() == 2);
+
+ CamelContext context = super.createCamelContext();
+ context.getRegistry().bind("hazelcastInstance", member);
+ return context;
+ }
+
+ private static Config memberConfig(String clusterName) {
+ Config config = new Config();
+ config.setClusterName(clusterName);
+ config.getMetricsConfig().setEnabled(false);
+
config.getNetworkConfig().getJoin().getMulticastConfig().setEnabled(false);
+
config.getNetworkConfig().getJoin().getAutoDetectionConfig().setEnabled(false);
+
config.getNetworkConfig().getJoin().getTcpIpConfig().setEnabled(true).addMember("127.0.0.1");
+ return config;
+ }
+
+ @AfterEach
+ void shutdownMembers() {
+ // JUnit runs this before the extension of CamelTestSupport stops the
context: stop it first, so that the
+ // consumers do not keep polling members that are shut down
+ context.stop();
+ if (member != null) {
+ member.shutdown();
+ }
+ if (otherMember != null) {
+ otherMember.shutdown();
+ }
+ }
+
+ @Test
+ void testCompletedExchangesAreCommitted() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:done");
+ mock.expectedBodiesReceivedInAnyOrder("a", "b", "c");
+
+ member.getQueue("done").offer("a");
+ member.getQueue("done").offer("b");
+ member.getQueue("done").offer("c");
+
+ mock.assertIsSatisfied();
+
+ // the application leaves the cluster (a restart or a redeploy): the
other member rolls back the transactions
+ // the application left open, which puts their messages back into the
queue
+ context.stop();
+ member.shutdown();
+ member = null;
+
+ await().during(3, TimeUnit.SECONDS).atMost(10, TimeUnit.SECONDS)
+ .untilAsserted(() -> assertEquals(0,
otherMember.getQueue("done").size(),
+ "The processed messages were not removed from the
queue"));
+ }
+
+ @Test
+ void testFailedExchangeIsRolledBack() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:failed");
+ mock.expectedBodiesReceived("x");
+
+ member.getQueue("failed").offer("x");
+
+ // the first attempt fails and must be rolled back, so the message is
taken again
+ mock.assertIsSatisfied();
+ assertEquals(2, attempts.get());
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("hazelcast-seda:done?transacted=true&hazelcastInstance=#hazelcastInstance").routeId("done")
+ .threads(1)
+ .to("mock:done");
+
+
from("hazelcast-seda:failed?transacted=true&hazelcastInstance=#hazelcastInstance").routeId("failed")
+ .threads(1)
+ .process(exchange -> {
+ if (attempts.incrementAndGet() == 1) {
+ throw new IllegalStateException("Forced
failure of the first attempt");
+ }
+ })
+ .to("mock:failed");
+ }
+ };
+ }
+}
diff --git
a/components/camel-hazelcast/src/test/java/org/apache/camel/component/hazelcast/HazelcastSedaTransactedPooledExchangeTest.java
b/components/camel-hazelcast/src/test/java/org/apache/camel/component/hazelcast/HazelcastSedaTransactedPooledExchangeTest.java
new file mode 100644
index 000000000000..d938bb5b8955
--- /dev/null
+++
b/components/camel-hazelcast/src/test/java/org/apache/camel/component/hazelcast/HazelcastSedaTransactedPooledExchangeTest.java
@@ -0,0 +1,34 @@
+/*
+ * 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.hazelcast;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.impl.engine.PooledExchangeFactory;
+
+/**
+ * The same transacted routes with pooled exchanges: a pooled exchange is
reset when it is done, so the consumer must
+ * read the outcome of the exchange before it is released, or a failed
exchange would be committed.
+ */
+class HazelcastSedaTransactedPooledExchangeTest extends
HazelcastSedaTransactedAsyncRouteTest {
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+ context.getCamelContextExtension().setExchangeFactory(new
PooledExchangeFactory());
+ return context;
+ }
+}
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 5587787d422e..94e1f7e5073f 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
@@ -945,6 +945,20 @@ kept apart. A valid token that has none of these claims is
now rejected with HTT
* The `CamelOAuthTokenValidationResult` exchange property is now set on the
exchange that the route processes, as
documented. Before, it was only set on the incoming HTTP exchange.
+=== camel-hazelcast - the transacted SEDA consumer waits for an exchange that
continues on another thread
+
+A Hazelcast transaction can only be committed or rolled back by the thread
that began it. When the route of a
+transacted (`transacted=true`) Hazelcast SEDA consumer continued on another
thread (an asynchronous producer, or
+`threads()`), the consumer tried to commit or roll back the transaction on
that thread, which failed with
+`Transaction cannot span multiple threads!` (logged as a WARN and ignored).
The message was neither removed nor
+put back: a failed message was not taken again, and the messages, also the
processed ones, were put back into the
+queue when the application left the Hazelcast cluster, so they were consumed
again. Now the consumer thread waits for
+the exchange to complete and ends the transaction itself, so it takes the next
message only then; use
+`concurrentConsumers` to process several messages at the same time. The
exchange must complete within the Hazelcast
+transaction timeout (2 minutes by default, `TransactionOptions`): when it
takes longer, the commit fails with
+`TransactionTimedOutException`, which is logged at ERROR, the transaction is
rolled back and the message is consumed
+again, also when the route succeeded.
+
=== camel-sql, camel-mybatis, camel-jooq - an exchange marked rollback only
last is not consumed either
The SQL, MyBatis and jOOQ consumers already left a message unconsumed when the
route marked the exchange rollback
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..75ae9dc56bd9 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
@@ -244,7 +244,13 @@ public interface HazelcastSedaEndpointBuilderFactory {
/**
* If set to true then the consumer runs in transaction mode, where the
* messages in the seda queue will only be removed if the transaction
- * commits, which happens when the processing is complete.
+ * commits, which happens when the processing is complete. A Hazelcast
+ * transaction belongs to the thread that began it, so each consumer
+ * thread waits for its exchange to complete, also when the route
+ * continues on another thread, before it takes the next message. The
+ * exchange must complete within the Hazelcast transaction timeout (2
+ * minutes by default): when it takes longer, the commit fails, the
+ * transaction is rolled back and the message is consumed again.
*
* The option is a: <code>boolean</code> type.
*
@@ -261,7 +267,13 @@ public interface HazelcastSedaEndpointBuilderFactory {
/**
* If set to true then the consumer runs in transaction mode, where the
* messages in the seda queue will only be removed if the transaction
- * commits, which happens when the processing is complete.
+ * commits, which happens when the processing is complete. A Hazelcast
+ * transaction belongs to the thread that began it, so each consumer
+ * thread waits for its exchange to complete, also when the route
+ * continues on another thread, before it takes the next message. The
+ * exchange must complete within the Hazelcast transaction timeout (2
+ * minutes by default): when it takes longer, the commit fails, the
+ * transaction is rolled back and the message is consumed again.
*
* The option will be converted to a <code>boolean</code> type.
*
@@ -647,7 +659,13 @@ public interface HazelcastSedaEndpointBuilderFactory {
/**
* If set to true then the consumer runs in transaction mode, where the
* messages in the seda queue will only be removed if the transaction
- * commits, which happens when the processing is complete.
+ * commits, which happens when the processing is complete. A Hazelcast
+ * transaction belongs to the thread that began it, so each consumer
+ * thread waits for its exchange to complete, also when the route
+ * continues on another thread, before it takes the next message. The
+ * exchange must complete within the Hazelcast transaction timeout (2
+ * minutes by default): when it takes longer, the commit fails, the
+ * transaction is rolled back and the message is consumed again.
*
* The option is a: <code>boolean</code> type.
*
@@ -664,7 +682,13 @@ public interface HazelcastSedaEndpointBuilderFactory {
/**
* If set to true then the consumer runs in transaction mode, where the
* messages in the seda queue will only be removed if the transaction
- * commits, which happens when the processing is complete.
+ * commits, which happens when the processing is complete. A Hazelcast
+ * transaction belongs to the thread that began it, so each consumer
+ * thread waits for its exchange to complete, also when the route
+ * continues on another thread, before it takes the next message. The
+ * exchange must complete within the Hazelcast transaction timeout (2
+ * minutes by default): when it takes longer, the commit fails, the
+ * transaction is rolled back and the message is consumed again.
*
* The option will be converted to a <code>boolean</code> type.
*
@@ -979,7 +1003,13 @@ public interface HazelcastSedaEndpointBuilderFactory {
/**
* If set to true then the consumer runs in transaction mode, where the
* messages in the seda queue will only be removed if the transaction
- * commits, which happens when the processing is complete.
+ * commits, which happens when the processing is complete. A Hazelcast
+ * transaction belongs to the thread that began it, so each consumer
+ * thread waits for its exchange to complete, also when the route
+ * continues on another thread, before it takes the next message. The
+ * exchange must complete within the Hazelcast transaction timeout (2
+ * minutes by default): when it takes longer, the commit fails, the
+ * transaction is rolled back and the message is consumed again.
*
* The option is a: <code>boolean</code> type.
*
@@ -996,7 +1026,13 @@ public interface HazelcastSedaEndpointBuilderFactory {
/**
* If set to true then the consumer runs in transaction mode, where the
* messages in the seda queue will only be removed if the transaction
- * commits, which happens when the processing is complete.
+ * commits, which happens when the processing is complete. A Hazelcast
+ * transaction belongs to the thread that began it, so each consumer
+ * thread waits for its exchange to complete, also when the route
+ * continues on another thread, before it takes the next message. The
+ * exchange must complete within the Hazelcast transaction timeout (2
+ * minutes by default): when it takes longer, the commit fails, the
+ * transaction is rolled back and the message is consumed again.
*
* The option will be converted to a <code>boolean</code> type.
*
