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 06f69193642a CAMEL-25527: camel-mybatis, camel-sql - do not run
onConsume for a failed or rollback only exchange (#27677)
06f69193642a is described below
commit 06f69193642a88ab7dab16d635620e435369efee
Author: allthingssecurity <[email protected]>
AuthorDate: Sun Oct 11 12:31:36 2026 +0530
CAMEL-25527: camel-mybatis, camel-sql - do not run onConsume for a failed
or rollback only exchange (#27677)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../camel/catalog/docs/mybatis-component.adoc | 5 +
.../src/main/docs/mybatis-component.adoc | 5 +
.../camel/component/mybatis/MyBatisConsumer.java | 10 +-
.../MyBatisOnConsumeFailedPooledExchangeTest.java | 34 ++++++
.../mybatis/MyBatisOnConsumeFailedTest.java | 129 +++++++++++++++++++++
.../apache/camel/component/sql/SqlConsumer.java | 20 ++--
.../SqlConsumerRollbackOnlyPooledExchangeTest.java | 34 ++++++
.../component/sql/SqlConsumerRollbackOnlyTest.java | 124 ++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_23.adoc | 9 ++
9 files changed, 359 insertions(+), 11 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/mybatis-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/mybatis-component.adoc
index e457330f334c..34c9939ba20b 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/mybatis-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/mybatis-component.adoc
@@ -607,6 +607,11 @@ And the statements in the sqlmap file:
</update>
----
+The statements only run when the exchange completed successfully. If the
exchange failed (or was marked rollback
+only), they are not run and the row is consumed again by the next poll, so a
row that always fails is consumed again at
+every poll. Handle the exception in the route (for example with
`onException(...).handled(true)`) to run the statements
+for a failed exchange anyway.
+
=== Participating in transactions
Setting up a transaction manager under camel-mybatis can be a little bit
diff --git a/components/camel-mybatis/src/main/docs/mybatis-component.adoc
b/components/camel-mybatis/src/main/docs/mybatis-component.adoc
index e457330f334c..34c9939ba20b 100644
--- a/components/camel-mybatis/src/main/docs/mybatis-component.adoc
+++ b/components/camel-mybatis/src/main/docs/mybatis-component.adoc
@@ -607,6 +607,11 @@ And the statements in the sqlmap file:
</update>
----
+The statements only run when the exchange completed successfully. If the
exchange failed (or was marked rollback
+only), they are not run and the row is consumed again by the next poll, so a
row that always fails is consumed again at
+every poll. Handle the exception in the route (for example with
`onException(...).handled(true)`) to run the statements
+for a failed exchange anyway.
+
=== Participating in transactions
Setting up a transaction manager under camel-mybatis can be a little bit
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 6bf0e4e35a32..6622f86c846a 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
@@ -136,18 +136,22 @@ public class MyBatisConsumer extends
ScheduledBatchPollingConsumer {
try {
getProcessor().process(exchange);
- if (onConsume != null) {
+ // 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()) {
endpoint.getProcessingStrategy().commit(endpoint,
exchange, data, onConsume);
}
} catch (Exception e) {
handleException(e);
}
- if (getEndpoint().isTransacted() && exchange.isFailed()) {
+ if (getEndpoint().isTransacted() && (exchange.isFailed() ||
exchange.isRollbackOnly())) {
// break out as we are transacted and should rollback
cause = exchange.getException();
if (cause == null) {
- cause = new RollbackExchangeException("Rollback
transaction due error processing exchange", null);
+ // rollback only, there was no error (the exception must
be created before the exchange is released)
+ cause = new RollbackExchangeException(
+ "Rollback transaction as the exchange was marked
as rollback only", exchange);
}
}
releaseExchange(exchange, false);
diff --git
a/components/camel-mybatis/src/test/java/org/apache/camel/component/mybatis/MyBatisOnConsumeFailedPooledExchangeTest.java
b/components/camel-mybatis/src/test/java/org/apache/camel/component/mybatis/MyBatisOnConsumeFailedPooledExchangeTest.java
new file mode 100644
index 000000000000..3da28b205295
--- /dev/null
+++
b/components/camel-mybatis/src/test/java/org/apache/camel/component/mybatis/MyBatisOnConsumeFailedPooledExchangeTest.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.mybatis;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.impl.engine.PooledExchangeFactory;
+
+/**
+ * The same cases with the pooled exchange factory, which resets an exchange
when it is released: the consumer must read
+ * the outcome of the exchange before it releases it.
+ */
+class MyBatisOnConsumeFailedPooledExchangeTest extends
MyBatisOnConsumeFailedTest {
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext camelContext = super.createCamelContext();
+ camelContext.getCamelContextExtension().setExchangeFactory(new
PooledExchangeFactory());
+ return camelContext;
+ }
+}
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
new file mode 100644
index 000000000000..0731a9656bb7
--- /dev/null
+++
b/components/camel-mybatis/src/test/java/org/apache/camel/component/mybatis/MyBatisOnConsumeFailedTest.java
@@ -0,0 +1,129 @@
+/*
+ * 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.mybatis;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.camel.RollbackExchangeException;
+import org.apache.camel.builder.RouteBuilder;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The onConsume statements must only run for a row whose exchange was
processed successfully: a failed or rollback only
+ * exchange must leave the row unprocessed, so the next poll consumes it again.
+ */
+class MyBatisOnConsumeFailedTest extends MyBatisTestSupport {
+
+ @Override
+ protected boolean createTestData() {
+ return false;
+ }
+
+ @Override
+ protected String getCreateStatement() {
+ return "create table ACCOUNT (ACC_ID INTEGER, ACC_FIRST_NAME
VARCHAR(255), ACC_LAST_NAME VARCHAR(255), ACC_EMAIL VARCHAR(255), PROCESSED
BOOLEAN DEFAULT false)";
+ }
+
+ @Test
+ void testFailedExchangeIsNotConsumed() throws Exception {
+ insertAccounts(account(1, "ok"), account(2, "fail"));
+
+ MyBatisConsumer consumer = (MyBatisConsumer)
context.getRoute("on-consume").getConsumer();
+ // one poll in the test thread (the scheduler is not started)
+ assertEquals(2, consumer.poll());
+
+ assertEquals(List.of(2), unprocessedAccountIds(), "only the
successfully processed account must be consumed");
+ }
+
+ @Test
+ void testRollbackOnlyExchangeIsNotConsumed() throws Exception {
+ insertAccounts(account(1, "ok"), account(2, "rollback"));
+
+ 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 must not
be consumed");
+ }
+
+ @Test
+ void testTransactedRollbackOnlyExchangeBreaksOutOfBatch() throws Exception
{
+ insertAccounts(account(1, "ok"), account(2, "rollback"), account(3,
"ok"));
+
+ MyBatisConsumer consumer = (MyBatisConsumer)
context.getRoute("transacted").getConsumer();
+ RollbackExchangeException e =
assertThrows(RollbackExchangeException.class, consumer::poll,
+ "a rollback only exchange must break out of a transacted
batch, as a failed exchange does");
+ // there was no error, and the message (logged by the consumer at
every such poll) must not say so
+ 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 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);
+ }
+
+ private static Account account(int id, String lastName) {
+ Account account = new Account();
+ account.setId(id);
+ account.setFirstName("Account" + id);
+ account.setLastName(lastName);
+ account.setEmailAddress("account" + id + "@example.com");
+ return account;
+ }
+
+ private List<Integer> unprocessedAccountIds() {
+ List<?> accounts =
template.requestBody("mybatis:selectUnprocessedAccounts?statementType=SelectList",
null,
+ List.class);
+ List<Integer> ids = new ArrayList<>();
+ for (Object account : accounts) {
+ ids.add(((Account) account).getId());
+ }
+ return ids;
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ public void configure() {
+
consume("mybatis:selectUnprocessedAccounts?onConsume=consumeAccount&startScheduler=false",
"on-consume");
+
consume("mybatis:selectUnprocessedAccounts?onConsume=consumeAccount&transacted=true&startScheduler=false",
+ "transacted");
+ }
+
+ private void consume(String uri, String routeId) {
+ from(uri).routeId(routeId)
+ .choice().when(simple("${body.lastName} ==
'rollback'")).markRollbackOnly().end()
+ .process(exchange -> {
+ Account account =
exchange.getIn().getBody(Account.class);
+ if ("fail".equals(account.getLastName())) {
+ throw new IllegalStateException("Simulated
failure for account " + account.getId());
+ }
+ })
+ .to("mock:processed");
+ }
+ };
+ }
+}
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 2a5c732dc4f3..dc03eb13005b 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,20 +278,24 @@ public class SqlConsumer extends
ScheduledBatchPollingConsumer {
exchange.setException(e);
}
- if (getEndpoint().isTransacted() && exchange.isFailed()) {
+ // a rollback only exchange (without an exception) did not
complete successfully either
+ boolean failed = exchange.isFailed() || exchange.isRollbackOnly();
+
+ if (getEndpoint().isTransacted() && failed) {
// break out as we are transacted and should rollback
Exception cause = exchange.getException();
+ if (cause == null) {
+ // rollback only, there was no error (the exception must
be created before the exchange is released)
+ cause = new RollbackExchangeException(
+ "Rollback transaction as the exchange was marked
as rollback only", exchange);
+ }
// must release exchange
releaseExchange(exchange, false);
- if (cause != null) {
- throw cause;
- } else {
- throw new RollbackExchangeException("Rollback transaction
due error processing exchange", null);
- }
+ throw cause;
}
// pick the on consume to use
- String sql = exchange.isFailed() ? onConsumeFailed : onConsume;
+ String sql = failed ? onConsumeFailed : onConsume;
try {
// we can only run on consume if there was data
if (data != null && sql != null) {
@@ -360,7 +364,7 @@ public class SqlConsumer extends
ScheduledBatchPollingConsumer {
}
/**
- * Sets a SQL to execute when the row failed being processed.
+ * Sets a SQL to execute when the row failed being processed (the exchange
failed or was marked rollback only).
*/
public void setOnConsumeFailed(String onConsumeFailed) {
this.onConsumeFailed = onConsumeFailed;
diff --git
a/components/camel-sql/src/test/java/org/apache/camel/component/sql/SqlConsumerRollbackOnlyPooledExchangeTest.java
b/components/camel-sql/src/test/java/org/apache/camel/component/sql/SqlConsumerRollbackOnlyPooledExchangeTest.java
new file mode 100644
index 000000000000..3164de8b51b4
--- /dev/null
+++
b/components/camel-sql/src/test/java/org/apache/camel/component/sql/SqlConsumerRollbackOnlyPooledExchangeTest.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.sql;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.impl.engine.PooledExchangeFactory;
+
+/**
+ * The same cases with the pooled exchange factory, which resets an exchange
when it is released: the consumer must read
+ * the outcome of the exchange before it releases it.
+ */
+class SqlConsumerRollbackOnlyPooledExchangeTest extends
SqlConsumerRollbackOnlyTest {
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext camelContext = super.createCamelContext();
+ camelContext.getCamelContextExtension().setExchangeFactory(new
PooledExchangeFactory());
+ return camelContext;
+ }
+}
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
new file mode 100644
index 000000000000..e125ab09c4ca
--- /dev/null
+++
b/components/camel-sql/src/test/java/org/apache/camel/component/sql/SqlConsumerRollbackOnlyTest.java
@@ -0,0 +1,124 @@
+/*
+ * 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.sql;
+
+import java.util.List;
+
+import org.apache.camel.RollbackExchangeException;
+import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.Test;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * A row whose exchange was marked rollback only (without an exception) did
not complete successfully: it must get
+ * onConsumeFailed instead of onConsume (no statement when onConsumeFailed is
not set), and it must break out of a
+ * transacted batch as a failed exchange does.
+ */
+class SqlConsumerRollbackOnlyTest extends CamelTestSupport {
+
+ private static final String ON_CONSUME = "&onConsume=update projects set
license = 'DONE' where id = :#id";
+ private static final String ON_CONSUME_FAILED = "&onConsumeFailed=update
projects set license = 'BAD' where id = :#id";
+
+ private EmbeddedDatabase db;
+ private JdbcTemplate jdbcTemplate;
+
+ @Override
+ public void doPreSetup() throws Exception {
+ db = new EmbeddedDatabaseBuilder()
+ .setName(getClass().getSimpleName())
+ .setType(EmbeddedDatabaseType.H2)
+ .addScript("sql/createAndPopulateDatabase.sql").build();
+
+ jdbcTemplate = new JdbcTemplate(db);
+ }
+
+ @Override
+ public void doPostTearDown() throws Exception {
+ if (db != null) {
+ db.shutdown();
+ }
+ }
+
+ @Test
+ void testRollbackOnlyExchangeRunsOnConsumeFailed() throws Exception {
+ SqlConsumer consumer = (SqlConsumer)
context.getRoute("on-consume").getConsumer();
+ // one poll in the test thread (the scheduler is not started)
+ assertEquals(3, consumer.poll());
+
+ assertEquals(List.of("DONE", "BAD", "DONE"), licenses(),
+ "the AMQ row, whose exchange was marked rollback only, must
get onConsumeFailed");
+ }
+
+ @Test
+ void testRollbackOnlyExchangeWithoutOnConsumeFailedIsNotConsumed() throws
Exception {
+ SqlConsumer consumer = (SqlConsumer)
context.getRoute("on-consume-only").getConsumer();
+ assertEquals(3, consumer.poll());
+
+ assertEquals(List.of("DONE", "ASF", "DONE"), licenses(),
+ "without onConsumeFailed, the AMQ row, whose exchange was
marked rollback only, must not get onConsume");
+ }
+
+ @Test
+ void testTransactedRollbackOnlyExchangeBreaksOutOfBatch() throws Exception
{
+ SqlConsumer consumer = (SqlConsumer)
context.getRoute("transacted").getConsumer();
+ // the consumer wraps the exception of the batch
+ RuntimeCamelException e = assertThrows(RuntimeCamelException.class,
consumer::poll,
+ "a rollback only exchange must break out of a transacted
batch, as a failed exchange does");
+ assertInstanceOf(RollbackExchangeException.class, e.getCause());
+ // there was no error, and the message (logged by the consumer at
every such poll) must not say so
+ assertTrue(e.getCause().getMessage().startsWith("Rollback transaction
as the exchange was marked as rollback only"),
+ e.getCause().getMessage());
+
+ assertEquals(List.of("DONE", "ASF", "XXX"), licenses(),
+ "neither the rollback only 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);
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ getContext().getComponent("sql",
SqlComponent.class).setDataSource(db);
+
+ consume("on-consume", ON_CONSUME + ON_CONSUME_FAILED);
+ consume("on-consume-only", ON_CONSUME);
+ consume("transacted", ON_CONSUME + ON_CONSUME_FAILED +
"&transacted=true");
+ }
+
+ private void consume(String routeId, String options) {
+ from("sql:select * from projects order by
id?startScheduler=false" + options)
+ .routeId(routeId)
+ .choice().when(simple("${body[PROJECT]} ==
'AMQ'")).markRollbackOnly().end()
+ .to("mock:processed");
+ }
+ };
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
index c2568bcd695c..ef55a568c050 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
@@ -863,6 +863,15 @@ comma and the statement ran once for each part (once for
each byte of a `byte[]`
Collections, iterators and other arrays (including `int[]`, `long[]` and other
primitive arrays) still run the statement
once per element; send a `List` to run it once per value.
+=== camel-mybatis, camel-sql - onConsume does not run for a failed or rollback
only exchange (Breaking change)
+
+The camel-mybatis consumer no longer runs the `onConsume` statements for a row
whose exchange failed or was marked
+rollback only: the row is consumed again by the next poll, so a row that
always fails is now processed at every poll.
+Handle the exception in the route (for example with
`onException(...).handled(true)`) to consume the row anyway. The
+camel-sql consumer now runs `onConsumeFailed` instead of `onConsume` for a
rollback only exchange (no statement when
+`onConsumeFailed` is not set, so the row is consumed again by the next poll).
With `transacted=true` both consumers now
+break out of the batch at a rollback only exchange, as they do at a failed one.
+
=== camel-groovy
A `GroovyShellFactory` is now looked up in the registry once per
`CamelContext`, when the first groovy expression is
