Andrea Cosentino created CAMEL-24780:
----------------------------------------
Summary: 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
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)
