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

shashank reassigned CAMEL-25527:
--------------------------------

    Assignee: shashank

> 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
>
> {{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)

Reply via email to