[
https://issues.apache.org/jira/browse/CAMEL-24780?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18116383#comment-18116383
]
Andrea Cosentino commented on CAMEL-24780:
------------------------------------------
Fixed via #26523 (squash-merged to main, 4.23.0).
_Claude Code on behalf of Andrea Cosentino (@oscerd)_
> camel-kafka - transactional producer leaks a half-begun transaction and skips
> the async callback when beginTransaction fails
> ----------------------------------------------------------------------------------------------------------------------------
>
> Key: CAMEL-24780
> URL: https://issues.apache.org/jira/browse/CAMEL-24780
> Project: Camel
> Issue Type: Bug
> Components: camel-kafka
> Reporter: Andrea Cosentino
> Assignee: Andrea Cosentino
> Priority: Major
> Fix For: 4.23.0
>
>
> In the asynchronous producer path, transactional set-up runs *before* the
> try/catch and in an order that can leave the UnitOfWork in an inconsistent
> state.
> h3. 1. Transaction begin (and body resolution) run outside the try
> {{KafkaProducer.process(Exchange, AsyncCallback)}} resolves
> {{message.getBody()}} and calls {{startKafkaTransaction(exchange)}} before
> the {{try}} block that guards sending. If
> {{kafkaProducer.beginTransaction()}} throws (or a lazy body conversion
> throws), the exception escapes {{process()}} *without calling*
> {{callback.done(...)}}, violating the async producer contract (the caller's
> callback is never completed).
> h3. 2. UnitOfWork is marked transacted before the transaction actually begins
> In {{startKafkaTransaction}} the order is
> {{uow.beginTransactedBy(transactionId)}} ->
> {{kafkaProducer.beginTransaction()}} -> {{uow.addSynchronization(...)}}. If
> {{beginTransaction()}} throws, the UoW is already flagged as transacted by
> this id but *no synchronization is registered*, so nothing ever
> commits/aborts and the flag dangles for the rest of the exchange.
> h3. Fix
> * Move the body resolution and {{startKafkaTransaction}} call inside the
> {{try}} so a failure is handled like any other send failure (set the
> exception and complete the async callback).
> * In {{startKafkaTransaction}}, call {{kafkaProducer.beginTransaction()}}
> first and only then {{uow.beginTransactedBy(transactionId)}} and
> {{uow.addSynchronization(...)}}, so a begin failure leaves the UoW untouched.
> Both changes are limited to the transactional async path; non-transactional
> sends are unaffected.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
