[jira] [Resolved] (CAMEL-25527) 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

["Claus Ibsen (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/CAMEL-25527?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Claus Ibsen resolved CAMEL-25527.
---------------------------------
    Resolution: Fixed

Fixed by https://github.com/apache/camel/pull/27677

> 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
>            Assignee: shashank
>            Priority: Major
>             Fix For: 4.23.0
>
>
> {{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 12 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.testRollbackOnlyExchangeWithoutOnConsumeFailedIsNotConsumed: without onConsumeFailed, the AMQ row, whose exchange was marked rollback only, must not get onConsume ==> expected: <[DONE, ASF, DONE]> but was: <[DONE, DONE, DONE]>
> SqlConsumerRollbackOnlyTest.testTransactedRollbackOnlyExchangeBreaksOutOfBatch: ... ==> Expected org.apache.camel.RuntimeCamelException to be thrown, but nothing was thrown.
> {noformat}
> (and the same six 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}} (no statement when {{onConsumeFailed}} is not set, as for a failed exchange). 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 323 (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)