[jira] [Work started] (CAMEL-24780) camel-kafka - transactional producer leaks a half-begun transaction and skips the async callback when beginTransaction fails

["Andrea Cosentino (Jira)" <ji...@apache.org>]

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

Work on CAMEL-24780 started by Andrea Cosentino.
------------------------------------------------
> 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
>
> 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)