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 ab5a934bd56d CAMEL-25588: camel-sql, camel-mybatis, camel-jooq - do
not consume an exchange marked rollback only last (#27883)
ab5a934bd56d is described below
commit ab5a934bd56d007e387c81b699e3ec4d5a51b640
Author: Andrea Cosentino <[email protected]>
AuthorDate: Thu Oct 15 14:00:18 2026 +0200
CAMEL-25588: camel-sql, camel-mybatis, camel-jooq - do not consume an
exchange marked rollback only last (#27883)
Co-Authored-By: Claude Opus 4.8 <[email protected]>
---
.../apache/camel/component/jooq/JooqConsumer.java | 3 +-
.../jooq/JooqConsumerDeleteFailedTest.java | 18 ++++++++++-
.../camel/component/mybatis/MyBatisConsumer.java | 7 ++--
.../mybatis/MyBatisOnConsumeFailedTest.java | 31 +++++++++++++++++-
.../apache/camel/component/sql/SqlConsumer.java | 5 +--
.../component/sql/SqlConsumerRollbackOnlyTest.java | 37 ++++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 11 +++++++
7 files changed, 104 insertions(+), 8 deletions(-)
diff --git
a/components/camel-jooq/src/main/java/org/apache/camel/component/jooq/JooqConsumer.java
b/components/camel-jooq/src/main/java/org/apache/camel/component/jooq/JooqConsumer.java
index bf1a46576f04..4f7c024583ec 100644
---
a/components/camel-jooq/src/main/java/org/apache/camel/component/jooq/JooqConsumer.java
+++
b/components/camel-jooq/src/main/java/org/apache/camel/component/jooq/JooqConsumer.java
@@ -123,7 +123,8 @@ public class JooqConsumer extends
ScheduledBatchPollingConsumer {
exchange.setException(e);
getExceptionHandler().handleException("Error processing
exchange", exchange, e);
}
- holder.consumed = !exchange.isFailed() &&
!exchange.isRollbackOnly();
+ // a rollback only (or rollback only last) exchange is not
consumed, so the next poll reads the row again
+ holder.consumed = !exchange.isFailed() &&
!exchange.isRollbackOnly() && !exchange.isRollbackOnlyLast();
releaseExchange(exchange, false);
}
diff --git
a/components/camel-jooq/src/test/java/org/apache/camel/component/jooq/JooqConsumerDeleteFailedTest.java
b/components/camel-jooq/src/test/java/org/apache/camel/component/jooq/JooqConsumerDeleteFailedTest.java
index 1bd7347c8f00..8b4c0f5e21ae 100644
---
a/components/camel-jooq/src/test/java/org/apache/camel/component/jooq/JooqConsumerDeleteFailedTest.java
+++
b/components/camel-jooq/src/test/java/org/apache/camel/component/jooq/JooqConsumerDeleteFailedTest.java
@@ -59,6 +59,19 @@ class JooqConsumerDeleteFailedTest extends BaseJooqTest {
assertEquals(List.of(2), remaining, "an author whose exchange was
marked rollback only must not be deleted");
}
+ @Test
+ void testRollbackOnlyLastExchangeKeepsRow() throws Exception {
+ create.executeInsert(new AuthorRecord(1, null, "ok", null, null,
null));
+ create.executeInsert(new AuthorRecord(2, null, "rollbackLast", null,
null, null));
+
+ JooqConsumer consumer = (JooqConsumer)
context.getRoute("delete-failed").getConsumer();
+ assertEquals(2, consumer.poll());
+
+ List<Integer> remaining =
create.select(AUTHOR.ID).from(AUTHOR).orderBy(AUTHOR.ID).fetch(AUTHOR.ID);
+ assertEquals(List.of(2), remaining,
+ "an author whose exchange was marked rollback only last must
not be deleted");
+ }
+
@Test
void testNotProcessedEntitiesKeepRows() throws Exception {
create.executeInsert(new AuthorRecord(1, null, "ok", null, null,
null));
@@ -85,7 +98,10 @@ class JooqConsumerDeleteFailedTest extends BaseJooqTest {
public void configure() {
from("jooq://org.apache.camel.component.jooq.db.tables.records.AuthorRecord?startScheduler=false")
.routeId("delete-failed")
- .choice().when(simple("${body.lastName} ==
'rollback'")).markRollbackOnly().end()
+ .choice()
+ .when(simple("${body.lastName} ==
'rollback'")).markRollbackOnly()
+ .when(simple("${body.lastName} ==
'rollbackLast'")).markRollbackOnlyLast()
+ .end()
.process(exchange -> {
AuthorRecord author =
exchange.getIn().getBody(AuthorRecord.class);
if ("fail".equals(author.getLastName())) {
diff --git
a/components/camel-mybatis/src/main/java/org/apache/camel/component/mybatis/MyBatisConsumer.java
b/components/camel-mybatis/src/main/java/org/apache/camel/component/mybatis/MyBatisConsumer.java
index 6622f86c846a..c58138ee372e 100644
---
a/components/camel-mybatis/src/main/java/org/apache/camel/component/mybatis/MyBatisConsumer.java
+++
b/components/camel-mybatis/src/main/java/org/apache/camel/component/mybatis/MyBatisConsumer.java
@@ -137,15 +137,16 @@ public class MyBatisConsumer extends
ScheduledBatchPollingConsumer {
getProcessor().process(exchange);
// only consume the data when the exchange was processed
successfully: a failed or rollback only
- // exchange leaves the data as it is, so the next poll
consumes it again
- if (onConsume != null && !exchange.isFailed() &&
!exchange.isRollbackOnly()) {
+ // exchange leaves the data as it is, so the next poll
consumes it again (rollback only last too)
+ if (onConsume != null && !exchange.isFailed() &&
!exchange.isRollbackOnly() && !exchange.isRollbackOnlyLast()) {
endpoint.getProcessingStrategy().commit(endpoint,
exchange, data, onConsume);
}
} catch (Exception e) {
handleException(e);
}
- if (getEndpoint().isTransacted() && (exchange.isFailed() ||
exchange.isRollbackOnly())) {
+ if (getEndpoint().isTransacted()
+ && (exchange.isFailed() || exchange.isRollbackOnly() ||
exchange.isRollbackOnlyLast())) {
// break out as we are transacted and should rollback
cause = exchange.getException();
if (cause == null) {
diff --git
a/components/camel-mybatis/src/test/java/org/apache/camel/component/mybatis/MyBatisOnConsumeFailedTest.java
b/components/camel-mybatis/src/test/java/org/apache/camel/component/mybatis/MyBatisOnConsumeFailedTest.java
index 0731a9656bb7..284ae2a9d620 100644
---
a/components/camel-mybatis/src/test/java/org/apache/camel/component/mybatis/MyBatisOnConsumeFailedTest.java
+++
b/components/camel-mybatis/src/test/java/org/apache/camel/component/mybatis/MyBatisOnConsumeFailedTest.java
@@ -65,6 +65,17 @@ class MyBatisOnConsumeFailedTest extends MyBatisTestSupport {
"an account whose exchange was marked rollback only must not
be consumed");
}
+ @Test
+ void testRollbackOnlyLastExchangeIsNotConsumed() throws Exception {
+ insertAccounts(account(1, "ok"), account(2, "rollbackLast"));
+
+ MyBatisConsumer consumer = (MyBatisConsumer)
context.getRoute("on-consume").getConsumer();
+ assertEquals(2, consumer.poll());
+
+ assertEquals(List.of(2), unprocessedAccountIds(),
+ "an account whose exchange was marked rollback only last must
not be consumed");
+ }
+
@Test
void testTransactedRollbackOnlyExchangeBreaksOutOfBatch() throws Exception
{
insertAccounts(account(1, "ok"), account(2, "rollback"), account(3,
"ok"));
@@ -81,6 +92,21 @@ class MyBatisOnConsumeFailedTest extends MyBatisTestSupport {
assertEquals(1,
getMockEndpoint("mock:processed").getReceivedCounter());
}
+ @Test
+ void testTransactedRollbackOnlyLastExchangeBreaksOutOfBatch() throws
Exception {
+ insertAccounts(account(1, "ok"), account(2, "rollbackLast"),
account(3, "ok"));
+
+ MyBatisConsumer consumer = (MyBatisConsumer)
context.getRoute("transacted").getConsumer();
+ RollbackExchangeException e =
assertThrows(RollbackExchangeException.class, consumer::poll,
+ "a rollback only last exchange must break out of a transacted
batch, as a failed exchange does");
+ assertTrue(e.getMessage().startsWith("Rollback transaction as the
exchange was marked as rollback only"),
+ e.getMessage());
+
+ assertEquals(List.of(2, 3), unprocessedAccountIds(),
+ "neither the rollback only last account nor the accounts after
it must be consumed");
+ assertEquals(1,
getMockEndpoint("mock:processed").getReceivedCounter());
+ }
+
private void insertAccounts(Account... accounts) {
template.sendBody("mybatis:insertAccount?statementType=Insert",
accounts);
}
@@ -115,7 +141,10 @@ class MyBatisOnConsumeFailedTest extends
MyBatisTestSupport {
private void consume(String uri, String routeId) {
from(uri).routeId(routeId)
- .choice().when(simple("${body.lastName} ==
'rollback'")).markRollbackOnly().end()
+ .choice()
+ .when(simple("${body.lastName} ==
'rollback'")).markRollbackOnly()
+ .when(simple("${body.lastName} ==
'rollbackLast'")).markRollbackOnlyLast()
+ .end()
.process(exchange -> {
Account account =
exchange.getIn().getBody(Account.class);
if ("fail".equals(account.getLastName())) {
diff --git
a/components/camel-sql/src/main/java/org/apache/camel/component/sql/SqlConsumer.java
b/components/camel-sql/src/main/java/org/apache/camel/component/sql/SqlConsumer.java
index dc03eb13005b..c96c2e80714b 100644
---
a/components/camel-sql/src/main/java/org/apache/camel/component/sql/SqlConsumer.java
+++
b/components/camel-sql/src/main/java/org/apache/camel/component/sql/SqlConsumer.java
@@ -278,8 +278,9 @@ public class SqlConsumer extends
ScheduledBatchPollingConsumer {
exchange.setException(e);
}
- // a rollback only exchange (without an exception) did not
complete successfully either
- boolean failed = exchange.isFailed() || exchange.isRollbackOnly();
+ // a rollback only exchange (without an exception) did not
complete successfully either; markRollbackOnlyLast
+ // reaches the consumer as the last transaction when no
transacted() EIP consumed the flag, so honor it too
+ boolean failed = exchange.isFailed() || exchange.isRollbackOnly()
|| exchange.isRollbackOnlyLast();
if (getEndpoint().isTransacted() && failed) {
// break out as we are transacted and should rollback
diff --git
a/components/camel-sql/src/test/java/org/apache/camel/component/sql/SqlConsumerRollbackOnlyTest.java
b/components/camel-sql/src/test/java/org/apache/camel/component/sql/SqlConsumerRollbackOnlyTest.java
index e125ab09c4ca..9a0addd3f358 100644
---
a/components/camel-sql/src/test/java/org/apache/camel/component/sql/SqlConsumerRollbackOnlyTest.java
+++
b/components/camel-sql/src/test/java/org/apache/camel/component/sql/SqlConsumerRollbackOnlyTest.java
@@ -82,6 +82,16 @@ class SqlConsumerRollbackOnlyTest extends CamelTestSupport {
"without onConsumeFailed, the AMQ row, whose exchange was
marked rollback only, must not get onConsume");
}
+ @Test
+ void testRollbackOnlyLastExchangeIsNotConsumed() throws Exception {
+ // CAMEL-25588: an exchange marked rollback only last must be treated
like rollback only - no onConsume
+ SqlConsumer consumer = (SqlConsumer)
context.getRoute("on-consume-only-last").getConsumer();
+ assertEquals(3, consumer.poll());
+
+ assertEquals(List.of("LAST", "ASF", "LAST"), licenses(),
+ "the AMQ row, whose exchange was marked rollback only last,
must not get onConsume");
+ }
+
@Test
void testTransactedRollbackOnlyExchangeBreaksOutOfBatch() throws Exception
{
SqlConsumer consumer = (SqlConsumer)
context.getRoute("transacted").getConsumer();
@@ -97,6 +107,20 @@ class SqlConsumerRollbackOnlyTest extends CamelTestSupport {
"neither the rollback only row nor the rows after it must be
consumed");
}
+ @Test
+ void testTransactedRollbackOnlyLastExchangeBreaksOutOfBatch() throws
Exception {
+ SqlConsumer consumer = (SqlConsumer)
context.getRoute("transacted-last").getConsumer();
+ // the consumer wraps the exception of the batch
+ RuntimeCamelException e = assertThrows(RuntimeCamelException.class,
consumer::poll,
+ "a rollback only last exchange must break out of a transacted
batch, as a failed exchange does");
+ assertInstanceOf(RollbackExchangeException.class, e.getCause());
+ assertTrue(e.getCause().getMessage().startsWith("Rollback transaction
as the exchange was marked as rollback only"),
+ e.getCause().getMessage());
+
+ assertEquals(List.of("LAST", "ASF", "XXX"), licenses(),
+ "neither the rollback only last row nor the rows after it must
be consumed");
+ }
+
private List<String> licenses() {
return jdbcTemplate.queryForList("select license from projects order
by id", String.class);
}
@@ -111,6 +135,19 @@ class SqlConsumerRollbackOnlyTest extends CamelTestSupport
{
consume("on-consume", ON_CONSUME + ON_CONSUME_FAILED);
consume("on-consume-only", ON_CONSUME);
consume("transacted", ON_CONSUME + ON_CONSUME_FAILED +
"&transacted=true");
+
+ // CAMEL-25588: a row marked rollback only last must not be
consumed either; a distinct onConsume
+ // (license = 'LAST') keeps this a separate endpoint from the
on-consume-only route above
+ from("sql:select * from projects order by
id?startScheduler=false&onConsume=update projects set license = 'LAST' where id
= :#id")
+ .routeId("on-consume-only-last")
+ .choice().when(simple("${body[PROJECT]} ==
'AMQ'")).markRollbackOnlyLast().end()
+ .to("mock:processed");
+
+ // CAMEL-25588: a transacted consumer must also break out of
the batch for a rollback only last row
+ from("sql:select * from projects order by
id?startScheduler=false&transacted=true&onConsume=update projects set license =
'LAST' where id = :#id")
+ .routeId("transacted-last")
+ .choice().when(simple("${body[PROJECT]} ==
'AMQ'")).markRollbackOnlyLast().end()
+ .to("mock:processed");
}
private void consume(String routeId, String options) {
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 260ff23a69d9..0942e3621447 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
@@ -212,6 +212,17 @@ A message that the route marks rollback only every time
was removed before, and
Handle such a message in the route instead, for example by moving it aside,
rather than marking it rollback only
every time.
+=== 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
+only with `markRollbackOnly()`. They now leave it unconsumed when the route
marks it rollback only last with
+`markRollbackOnlyLast()` too, as the JPA and Hazelcast SEDA consumers do: the
SQL consumer runs its `onConsumeFailed`
+statement instead of `onConsume` (and no statement when `onConsumeFailed` is
not set), the MyBatis consumer does not
+run its `onConsume` statement, and the jOOQ consumer does not delete the
entity, so the next poll reads the row
+again. With `transacted=true` the SQL and MyBatis consumers break out of the
batch and roll back.
+`markRollbackOnlyLast()` reaches the consumer only when no `transacted()` EIP
has consumed the flag, so the
+consumer's own transaction is the last one.
+
=== camel-telegram - webhook requests are verified with a secret token
In webhook mode, Camel now registers the webhook with a secret token, which
Telegram sends in the
