shashank created CAMEL-25527:
--------------------------------
Summary: camel-mybatis, camel-sql - the consumer runs onConsume
for a failed (camel-mybatis) or rollback only (both) exchange, so the row is
marked as consumed and not polled again
Key: CAMEL-25527
URL: https://issues.apache.org/jira/browse/CAMEL-25527
Project: Camel
Issue Type: Bug
Components: camel-mybatis, camel-sql
Reporter: shashank
{{MyBatisConsumer.processBatch}} (lines 135-153 at main 586855a18315) runs the
{{onConsume}} statements right after {{getProcessor().process(exchange)}},
whatever happened to the exchange:
{code:java}
getProcessor().process(exchange);
if (onConsume != null) {
endpoint.getProcessingStrategy().commit(endpoint, exchange, data,
onConsume);
}
{code}
A failed route does not throw out of {{process}}; the error handler leaves the
exception on the exchange. So the row of a failed exchange is marked as
processed (in the documented example {{update ACCOUNT set PROCESSED = true}})
and the next poll does not pick it up again: the row is lost from the "queue".
The same happens for an exchange marked rollback only. The javadoc of
{{MyBatisConsumer.setOnConsume}} says the statements run "after successful
processing", and the component docs use {{onConsume}} so that a row is not
consumed twice.
{{SqlConsumer.processBatch}} (line 294) picks {{onConsumeFailed}} from
{{exchange.isFailed()}}, which is only {{exception != null}}. An exchange
marked rollback only (for example with {{markRollbackOnly()}}) has no
exception, so it gets {{onConsume}}, the query for a row that "was processed
successfully". With {{transacted=true}} both consumers break out of the batch
only when {{exchange.isFailed()}} (lines 146 and 281), so a rollback only
exchange does not stop a transacted batch either. The code for that case ({{new
RollbackExchangeException(..., null)}}) has never run: it would throw a
{{NullPointerException}} because the exception requires the exchange.
h3. Reproduction
New {{MyBatisOnConsumeFailedTest}} (H2, as the other camel-mybatis tests) and
{{SqlConsumerRollbackOnlyTest}} (embedded H2, as
{{SqlConsumerDeleteFailedTest}}). The routes use {{startScheduler=false}} and
the tests call {{poll()}} once, so they are deterministic. Each class has a
subclass that runs the same tests with {{PooledExchangeFactory}}. On main all
10 tests fail (two runs):
{noformat}
MyBatisOnConsumeFailedTest.testFailedExchangeIsNotConsumed: only the
successfully processed account must be consumed ==> expected: <[2]> but was:
<[]>
MyBatisOnConsumeFailedTest.testRollbackOnlyExchangeIsNotConsumed: an account
whose exchange was marked rollback only must not be consumed ==> expected:
<[2]> but was: <[]>
MyBatisOnConsumeFailedTest.testTransactedRollbackOnlyExchangeBreaksOutOfBatch:
... ==> Expected org.apache.camel.RollbackExchangeException to be thrown, but
nothing was thrown.
SqlConsumerRollbackOnlyTest.testRollbackOnlyExchangeRunsOnConsumeFailed: the
AMQ row, whose exchange was marked rollback only, must get onConsumeFailed ==>
expected: <[DONE, BAD, DONE]> but was: <[DONE, DONE, DONE]>
SqlConsumerRollbackOnlyTest.testTransactedRollbackOnlyExchangeBreaksOutOfBatch:
... ==> Expected org.apache.camel.RuntimeCamelException to be thrown, but
nothing was thrown.
{noformat}
(and the same five in the {{PooledExchangeFactory}} subclasses).
h3. Proposed fix
camel-mybatis runs {{onConsume}} only when the exchange is neither failed nor
rollback only; camel-sql treats a rollback only exchange as failed when it
picks {{onConsume}} or {{onConsumeFailed}}. With {{transacted=true}} both
consumers break out of the batch at a rollback only exchange as they do at a
failed one, and they create the {{RollbackExchangeException}} with the
exchange, before releasing it. Both consumers already create their exchanges
with {{createExchange(false)}} and release them after reading the outcome, so
the change also holds with the pooled exchange factory (the pooled tests check
this). The camel-mybatis docs (and the catalog copy) say that a row whose
exchange failed is consumed again by the next poll and how to consume it anyway
({{onException(...).handled(true)}}); the 4.23 upgrade guide has a note. Tests
with the fix: camel-mybatis 52, camel-sql 321 (3 skipped), 0 failures;
{{SqlFunctionDataSourceTest}} needs MariaDB4j, which does not install on the
test machine, and fails the same way on main.
Found while checking the siblings of the camel-jooq fix (CAMEL-25511), the
other consumers that use a table as a queue, against the property of its Lean 4
model, "a row is consumed only if its exchange succeeded" (by reading the code;
no new model). Confirmed with the real consumers by the tests above.
Affected: main, camel-4.22.x, camel-4.18.x, camel-4.14.x (same code, GitHub
contents API; latest releases 4.22.1, 4.18.4, 4.14.9).
Duplicate check (2026-10-10): JIRA text "onConsume" + "mybatis" (2: CAMEL-19746
docs examples, CAMEL-5485 transaction demarcation when onConsume fails, both
different), "onConsume" + "failed" (2: CAMEL-25511 jooq, CAMEL-14337
unrelated), "onConsumeFailed" (1: CAMEL-25511), "MyBatisConsumer" (none),
"SqlConsumer" + "rollback" (none), "markRollbackOnly" in
camel-sql/camel-mybatis (none); GitHub pull requests "mybatis onConsume",
"SqlConsumer rollback", "sql onConsumeFailed", "onConsume failed", "mybatis
consumer failed": none ("MyBatisConsumer" finds only #27644, the jooq fix). No
open pull request changes {{MyBatisConsumer}} or {{SqlConsumer}}.
_Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
