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 7f3ea874a499 CAMEL-25550: camel-jpa, camel-hazelcast - a transacted
consumer must not commit an exchange marked rollback only (#27681)
7f3ea874a499 is described below
commit 7f3ea874a4994eac62dd1a537efeba9c0646650b
Author: Claus Ibsen <[email protected]>
AuthorDate: Mon Oct 12 17:08:07 2026 +0200
CAMEL-25550: camel-jpa, camel-hazelcast - a transacted consumer must not
commit an exchange marked rollback only (#27681)
* CAMEL-25550: camel-jpa, camel-hazelcast - a transacted consumer must not
commit an exchange marked rollback only
An exchange marked rollback only (markRollbackOnly(),
markRollbackOnlyLast()) has no exception, so the JPA consumer deleted the
entity (or called @Consumed) and committed the poll, and the transacted
Hazelcast SEDA consumer committed its transaction, removing the message from
the queue. The JPA consumer now does not consume such an entity: with
transacted=true it throws a RollbackExchangeException so the whole poll is
rolled back, with transacted=false it skips the entity and continue [...]
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
Signed-off-by: Claus Ibsen <[email protected]>
* CAMEL-25550: Address review feedback
Fix the grammar of the rollback only exception message, and move the
upgrade guide note from 4.23 to 4.24 as this does not go into 4.23.0.
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
Signed-off-by: Claus Ibsen <[email protected]>
* CAMEL-25550: the docs say an entity is consumed again only while it
matches the query, and what a message always rolled back does now
With transacted=false a change made by @PreConsumed is committed with the
poll, so an entity it moved out of the
query is not polled again. The upgrade note also says that a message always
marked rollback only, which was removed
before, is now consumed again at every poll (JPA) or taken again at once
(transacted Hazelcast SEDA, see CAMEL-25589).
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_01STT6whBgK1AqsSsUKrnE8m
---------
Signed-off-by: Claus Ibsen <[email protected]>
Co-authored-by: Claude Opus 5.5 (1M context) <[email protected]>
---
.../apache/camel/catalog/docs/jpa-component.adoc | 7 +
.../hazelcast/seda/HazelcastSedaConsumer.java | 13 +-
.../HazelcastSedaTransactedRollbackOnlyTest.java | 112 +++++++++++++++
.../camel-jpa/src/main/docs/jpa-component.adoc | 7 +
.../apache/camel/component/jpa/JpaConsumer.java | 17 +++
.../component/jpa/JpaConsumerRollbackOnlyTest.java | 155 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 16 +++
7 files changed, 322 insertions(+), 5 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/jpa-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/jpa-component.adoc
index a61fe52350fc..72d7b1e71ffb 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/jpa-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/jpa-component.adoc
@@ -92,6 +92,13 @@ processed (such as to exclude it from a future query), then
you can
annotate a method with `@Consumed` (org.apache.camel.component.jpa.Consumed).
It will be invoked on your entity bean when the entity bean has been processed
(and when routing is done).
+An entity bean is only deleted (or `@Consumed` invoked) when its exchange
completed successfully. If the route marks
+the exchange rollback only (for example with `markRollbackOnly()`), the entity
bean is not consumed and stays in the
+table, so the next poll consumes it again, as long as it still matches the
query. With `transacted=false` the other
+entity beans of the poll are still consumed, and so is what a `@PreConsumed`
method changed before routing: an entity
+bean that `@PreConsumed` moved out of the query is not polled again. With
`transacted=true` the whole poll is rolled
+back, the changes of `@PreConsumed` included.
+
You can use `@PreConsumed` (org.apache.camel.component.jpa.PreConsumed)
which will be invoked on your entity bean before it has been processed
(before routing).
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..a444a6c19d42 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
@@ -115,16 +115,19 @@ public class HazelcastSedaConsumer extends
DefaultConsumer implements Runnable {
// 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) {
- // Rollback
- if (txc != null) {
- txc.rollbackTransaction();
- }
getExceptionHandler().handleException("Error processing exchange", exchange,
exchange.getException());
}
// It's OK, I commit
- if (exchange.getException() == null && txc !=
null) {
+ 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/HazelcastSedaTransactedRollbackOnlyTest.java
b/components/camel-hazelcast/src/test/java/org/apache/camel/component/hazelcast/HazelcastSedaTransactedRollbackOnlyTest.java
new file mode 100644
index 000000000000..c92b55743aba
--- /dev/null
+++
b/components/camel-hazelcast/src/test/java/org/apache/camel/component/hazelcast/HazelcastSedaTransactedRollbackOnlyTest.java
@@ -0,0 +1,112 @@
+/*
+ * 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.concurrent.TimeUnit;
+
+import com.hazelcast.collection.IQueue;
+import com.hazelcast.core.HazelcastInstance;
+import com.hazelcast.transaction.TransactionContext;
+import com.hazelcast.transaction.TransactionalQueue;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+
+import static org.mockito.AdditionalAnswers.answersWithDelay;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.atLeastOnce;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * A message whose exchange was marked rollback only (without an exception)
must roll back the Hazelcast transaction, so
+ * the message is not removed from the queue.
+ */
+public class HazelcastSedaTransactedRollbackOnlyTest extends
HazelcastCamelTestSupport {
+
+ @Mock
+ private IQueue<Object> queue;
+
+ @Mock
+ private TransactionalQueue<Object> tqueue;
+
+ private final TransactionContext okTransaction =
Mockito.mock(TransactionContext.class);
+ private final TransactionContext kaboomTransaction =
Mockito.mock(TransactionContext.class);
+ private final TransactionContext idleTransaction =
Mockito.mock(TransactionContext.class);
+
+ @Override
+ protected void trainHazelcastInstance(HazelcastInstance hazelcastInstance)
{
+ // one transaction per polled message, and then one for each (empty)
poll after
+ when(hazelcastInstance.newTransactionContext())
+ .thenReturn(okTransaction, kaboomTransaction, idleTransaction);
+ when(hazelcastInstance.getQueue("foo")).thenReturn(queue);
+ for (TransactionContext transaction : new TransactionContext[] {
okTransaction, kaboomTransaction, idleTransaction }) {
+ when(transaction.getQueue("foo")).thenReturn(tqueue);
+ }
+ }
+
+ @Override
+ protected void verifyHazelcastInstance(HazelcastInstance
hazelcastInstance) {
+ verify(hazelcastInstance, atLeastOnce()).getQueue("foo");
+ verify(hazelcastInstance, atLeastOnce()).newTransactionContext();
+ }
+
+ @Test
+ public void testRollbackOnlyRollsBackTransaction() throws Exception {
+ when(tqueue.poll(anyLong(), any(TimeUnit.class)))
+ .thenReturn("ok")
+ .thenReturn("kaboom")
+ .thenAnswer(answersWithDelay(50, invocation -> null));
+
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("ok");
+
+ // start the consumer after the queue is trained
+ context.getRouteController().startRoute("foo");
+
+ MockEndpoint.assertIsSatisfied(context);
+
+ verify(kaboomTransaction, timeout(5000)).rollbackTransaction();
+ verify(kaboomTransaction, never()).commitTransaction();
+
+ verify(okTransaction).commitTransaction();
+ verify(okTransaction, never()).rollbackTransaction();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("hazelcast-seda:foo?transacted=true").routeId("foo").autoStartup(false)
+
.choice().when(body().isEqualTo("kaboom")).markRollbackOnly().end()
+ .to("mock:result");
+ }
+ };
+ }
+
+ @AfterEach
+ public final void stopContext() throws Exception {
+ context.stop();
+ }
+}
diff --git a/components/camel-jpa/src/main/docs/jpa-component.adoc
b/components/camel-jpa/src/main/docs/jpa-component.adoc
index a61fe52350fc..72d7b1e71ffb 100644
--- a/components/camel-jpa/src/main/docs/jpa-component.adoc
+++ b/components/camel-jpa/src/main/docs/jpa-component.adoc
@@ -92,6 +92,13 @@ processed (such as to exclude it from a future query), then
you can
annotate a method with `@Consumed` (org.apache.camel.component.jpa.Consumed).
It will be invoked on your entity bean when the entity bean has been processed
(and when routing is done).
+An entity bean is only deleted (or `@Consumed` invoked) when its exchange
completed successfully. If the route marks
+the exchange rollback only (for example with `markRollbackOnly()`), the entity
bean is not consumed and stays in the
+table, so the next poll consumes it again, as long as it still matches the
query. With `transacted=false` the other
+entity beans of the poll are still consumed, and so is what a `@PreConsumed`
method changed before routing: an entity
+bean that `@PreConsumed` moved out of the query is not polled again. With
`transacted=true` the whole poll is rolled
+back, the changes of `@PreConsumed` included.
+
You can use `@PreConsumed` (org.apache.camel.component.jpa.PreConsumed)
which will be invoked on your entity bean before it has been processed
(before routing).
diff --git
a/components/camel-jpa/src/main/java/org/apache/camel/component/jpa/JpaConsumer.java
b/components/camel-jpa/src/main/java/org/apache/camel/component/jpa/JpaConsumer.java
index e55c4bfc5f66..47d50cc691bd 100644
---
a/components/camel-jpa/src/main/java/org/apache/camel/component/jpa/JpaConsumer.java
+++
b/components/camel-jpa/src/main/java/org/apache/camel/component/jpa/JpaConsumer.java
@@ -36,6 +36,7 @@ import jakarta.persistence.Query;
import org.apache.camel.Exchange;
import org.apache.camel.ExchangePropertyKey;
import org.apache.camel.Processor;
+import org.apache.camel.RollbackExchangeException;
import org.apache.camel.support.ObjectHelper;
import org.apache.camel.support.ScheduledBatchPollingConsumer;
import org.apache.camel.util.AnnotationHelper;
@@ -210,6 +211,18 @@ public class JpaConsumer extends
ScheduledBatchPollingConsumer {
if (exchange.getException() != null) {
// if we failed then throw exception
throw exchange.getException();
+ } else if (exchange.isRollbackOnly() ||
exchange.isRollbackOnlyLast()) {
+ // the route marked the exchange rollback only (there
is no exception): the entity is not
+ // consumed, so the next poll consumes it again
+ if (isTransacted()) {
+ // throw to roll back the whole batch
+ // (the exception must be created before the
exchange is released)
+ throw new RollbackExchangeException(
+ "Rollback transaction due to exchange
being marked for rollback only", exchange);
+ }
+ // not transacted: a deliberate rollback is not an
error, so skip this entity and continue
+ // with the next one in the batch
+ LOG.debug("Exchange marked for rollback only so the
entity is not consumed: {}", exchange);
} else {
// Run the @Consumed callback
getDeleteHandler().deleteObject(batchEntityManager,
result, exchange);
@@ -325,6 +338,10 @@ public class JpaConsumer extends
ScheduledBatchPollingConsumer {
* This option is default <tt>false</tt>. When <tt>false</tt> then all the
good messages will commit, and the first
* failed message will rollback. However when <tt>true</tt>, then all
messages will rollback, if just one message
* failed.
+ * <p/>
+ * A message whose exchange was marked rollback only (for example with
<tt>markRollbackOnly()</tt>) is not consumed:
+ * when <tt>false</tt> the other messages of the batch are still
committed, when <tt>true</tt> all messages will
+ * rollback.
*/
public void setTransacted(boolean transacted) {
this.transacted = transacted;
diff --git
a/components/camel-jpa/src/test/java/org/apache/camel/component/jpa/JpaConsumerRollbackOnlyTest.java
b/components/camel-jpa/src/test/java/org/apache/camel/component/jpa/JpaConsumerRollbackOnlyTest.java
new file mode 100644
index 000000000000..83062f0bc4ee
--- /dev/null
+++
b/components/camel-jpa/src/test/java/org/apache/camel/component/jpa/JpaConsumerRollbackOnlyTest.java
@@ -0,0 +1,155 @@
+/*
+ * 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.jpa;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import jakarta.persistence.PersistenceException;
+
+import org.apache.camel.RollbackExchangeException;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.examples.MultiSteps;
+import org.apache.camel.examples.SendEmail;
+import org.apache.camel.processor.jpa.AbstractJpaTest;
+import org.apache.camel.util.ObjectHelper;
+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.assertThrows;
+
+/**
+ * An entity whose exchange was marked rollback only (without an exception)
must not be consumed: the consumer must not
+ * delete it (or call its @Consumed callback), and with transacted=true the
whole poll must roll back.
+ */
+public class JpaConsumerRollbackOnlyTest extends AbstractJpaTest {
+
+ private static final String SELECT_ALL_STRING = "select x from " +
SendEmail.class.getName() + " x";
+ private static final String KABOOM = "[email protected]";
+
+ @Test
+ public void testNonTransactedRollbackOnlyEntityIsNotDeleted() throws
Exception {
+ insertSendEmails("[email protected]", KABOOM, "[email protected]");
+
+ JpaConsumer consumer = (JpaConsumer)
context.getRoute("non-tx").getConsumer();
+ // one poll in the test thread (the scheduler is not started)
+ assertEquals(3, consumer.poll());
+
+ assertEquals(List.of(KABOOM), sendEmailAddresses(),
+ "only the entity whose exchange was marked rollback only must
stay in the table");
+ assertEquals(2, getMockEndpoint("mock:result").getReceivedCounter());
+
+ // the rollback only entity is consumed again by the next poll
+ assertEquals(1, consumer.poll());
+ assertEquals(List.of(KABOOM), sendEmailAddresses());
+ assertEquals(4, getMockEndpoint("mock:start").getReceivedCounter());
+ }
+
+ @Test
+ public void testTransactedRollbackOnlyEntityRollsBackThePoll() throws
Exception {
+ insertSendEmails("[email protected]", KABOOM, "[email protected]");
+
+ JpaConsumer consumer = (JpaConsumer)
context.getRoute("tx").getConsumer();
+ PersistenceException e = assertThrows(PersistenceException.class,
consumer::poll,
+ "a rollback only exchange must roll back a transacted poll");
+
assertNotNull(ObjectHelper.getException(RollbackExchangeException.class, e),
+ "the poll must be rolled back by a RollbackExchangeException");
+
+ assertEquals(List.of("[email protected]", "[email protected]", KABOOM),
sendEmailAddresses(),
+ "no entity must be deleted when the transacted poll is rolled
back");
+ }
+
+ @Test
+ public void testNonTransactedRollbackOnlyEntityIsNotConsumedCallback()
throws Exception {
+ for (String address : List.of("[email protected]", KABOOM,
"[email protected]")) {
+ saveEntityInDB(new MultiSteps(address));
+ }
+
+ JpaConsumer consumer = (JpaConsumer)
context.getRoute("consumed").getConsumer();
+ assertEquals(3, consumer.poll());
+
+ entityManager.clear();
+ List<?> step1 = entityManager.createQuery("select x from " +
MultiSteps.class.getName() + " x where x.step = 1")
+ .getResultList();
+ assertEquals(1, step1.size(), "the @Consumed callback must not be
called for the rollback only entity");
+ assertEquals(KABOOM, ((MultiSteps) step1.get(0)).getAddress());
+ }
+
+ private void insertSendEmails(String... addresses) {
+ for (String address : addresses) {
+ saveEntityInDB(new SendEmail(address));
+ }
+ }
+
+ private List<String> sendEmailAddresses() {
+ entityManager.clear();
+ List<?> list =
entityManager.createQuery(SELECT_ALL_STRING).getResultList();
+ List<String> answer = new ArrayList<>();
+ for (Object item : list) {
+ answer.add(((SendEmail) item).getAddress());
+ }
+ answer.sort(null);
+ return answer;
+ }
+
+ @Override
+ protected void cleanupRepository() {
+ super.cleanupRepository();
+ transactionTemplate.execute(status -> {
+ entityManager.joinTransaction();
+ List<?> list = entityManager.createQuery("select x from " +
MultiSteps.class.getName() + " x").getResultList();
+ for (Object item : list) {
+ entityManager.remove(item);
+ }
+ entityManager.flush();
+ return Boolean.TRUE;
+ });
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("jpa://" + SendEmail.class.getName() +
"?transacted=false&startScheduler=false").routeId("non-tx")
+ .to("mock:start")
+ .choice().when(simple("${body.address} == '" + KABOOM
+ "'")).markRollbackOnly().end()
+ .to("mock:result");
+
+ from("jpa://" + SendEmail.class.getName() +
"?transacted=true&startScheduler=false").routeId("tx")
+ .choice().when(simple("${body.address} == '" + KABOOM
+ "'")).markRollbackOnly().end()
+ .to("mock:result");
+
+ from("jpa://" + MultiSteps.class.getName() + "?query=select x
from " + MultiSteps.class.getName()
+ + " x where x.step =
1&startScheduler=false").routeId("consumed")
+ .choice().when(simple("${body.address} == '" + KABOOM
+ "'")).markRollbackOnly().end()
+ .to("mock:result");
+ }
+ };
+ }
+
+ @Override
+ protected String routeXml() {
+ return "org/apache/camel/processor/jpa/springJpaRouteTest.xml";
+ }
+
+ @Override
+ protected String selectAllString() {
+ return SELECT_ALL_STRING;
+ }
+}
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 87b8d1a543b6..24d8b1ea2530 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
@@ -29,6 +29,22 @@ that a route sends to and no route consumes, so they report
the same endpoints o
So `camel validate` can report an endpoint it did not report before.
+=== camel-jpa, camel-hazelcast - an exchange marked rollback only is not
consumed
+
+When a route marks the exchange rollback only (for example with
`markRollbackOnly()` or `markRollbackOnlyLast()`)
+there is no exception on the exchange, and these consumers treated it as a
success. The JPA consumer deleted the entity
+(or called its `@Consumed` method) and committed the poll, and the transacted
(`transacted=true`) Hazelcast SEDA
+consumer committed its transaction, so the message was removed from the queue.
Now a rollback only exchange is not
+consumed: the JPA consumer leaves the entity in the table, so the next poll
consumes it again, as long as it still
+matches the query. With `transacted=false` the other entities of the poll are
still consumed (without logging a WARN),
+and so is what a `@PreConsumed` method changed before routing; with
`transacted=true` the whole poll is rolled back,
+as for a failed exchange. The Hazelcast SEDA consumer rolls back its
transaction, so the message stays in the queue.
+
+A message that the route marks rollback only every time was removed before,
and is now consumed again at every poll
+(JPA), or taken from the queue again at once, without a delay (transacted
Hazelcast SEDA, as for a failed exchange).
+Handle such a message in the route instead, for example by moving it aside,
rather than marking it rollback only
+every time.
+
=== camel-hashicorp-vault - properties without engine use the `secret` engine
A `hashicorp:` property without an engine, such as `{{hashicorp:mysecret}}` or
`{{hashicorp:mysecret#field}}`, now
