[
https://issues.apache.org/jira/browse/CAMEL-25527?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Claus Ibsen updated CAMEL-25527:
--------------------------------
Fix Version/s: 4.23.0
> 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)
