gnodet-bot commented on code in PR #26523:
URL: https://github.com/apache/camel/pull/26523#discussion_r4028430702
##########
components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerTest.java:
##########
@@ -189,6 +192,39 @@ public void processAsyncSendsMessageWithException() {
assertRecordMetadataExists();
}
+ @Test
+ public void processAsyncCompletesCallbackWhenBeginTransactionFails()
throws Exception {
+ // CAMEL-24780: a failure to begin the transaction must set the
exception and complete the async
+ // callback rather than escaping process(), and it must not leave the
unit of work flagged as
+ // transacted without a synchronization to commit or roll it back.
+ setTransactionId(producer, "test-tx");
+ endpoint.getConfiguration().setTopic("sometopic");
+
+ Producer kp = producer.getKafkaProducer();
+ Mockito.doThrow(new ApiException("cannot
begin")).when(kp).beginTransaction();
+
+ UnitOfWork uow = Mockito.mock(UnitOfWork.class);
+ Mockito.when(uow.isTransactedBy(any())).thenReturn(false);
+ Mockito.when(exchange.getUnitOfWork()).thenReturn(uow);
+ Mockito.when(exchange.getIn()).thenReturn(in);
+ Mockito.when(exchange.getMessage()).thenReturn(in);
+
+ boolean sync = producer.process(exchange, callback);
+
+ assertTrue(sync);
+ Mockito.verify(exchange).setException(isA(ApiException.class));
+ Mockito.verify(callback).done(eq(true));
+ // begin failed, so the unit of work must be left untouched (no
dangling transacted flag)
+ Mockito.verify(uow, Mockito.never()).beginTransactedBy(any());
+ Mockito.verify(uow, Mockito.never()).addSynchronization(any());
+ }
Review Comment:
**Missing test for the "transaction started, send fails" path.**
The new test covers `beginTransaction()` throwing — good. But the other half
of Fix 1's correctness story is: when `beginTransaction()` **succeeds** and
then `doSend()` / `createRecord()` fails, the `KafkaTransactionSynchronization`
that was registered must fire `abortTransaction()` via the UoW lifecycle.
This path is handled by the existing Camel UoW mechanism (not by any code
introduced here), but a test asserting that `kafkaProducer.abortTransaction()`
is called when a send fails in transactional mode would provide a regression
guard for the full scenario and make the contract explicit.
##########
components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java:
##########
@@ -518,8 +521,11 @@ private void startKafkaTransaction(Exchange exchange) {
if (!uow.isTransactedBy(transactionId)) {
LOG.debug("Starting kafka transaction {} with exchange {}",
transactionId, exchange.getExchangeId());
- uow.beginTransactedBy(transactionId);
+ // Begin the broker transaction first, then mark the unit of work
and register the
+ // synchronization. This way a failure in beginTransaction() does
not leave the unit of work
+ // flagged as transacted without a synchronization to commit or
roll it back (CAMEL-24780).
kafkaProducer.beginTransaction();
+ uow.beginTransactedBy(transactionId);
uow.addSynchronization(new
KafkaTransactionSynchronization(transactionId, kafkaProducer));
Review Comment:
**Latent window between `beginTransaction()` and `addSynchronization()`.**
The reorder is correct for the `beginTransaction()` failure case. However,
if `beginTransaction()` succeeds and then anything between lines 528-529 threw
(e.g. an `OutOfMemoryError` or a misbehaving `UnitOfWork` implementation), the
Kafka transaction would be open with no synchronization registered to abort it.
In practice `DefaultUnitOfWork.beginTransactedBy()` and
`addSynchronization()` are in-memory ArrayList ops that won't throw, so this is
theoretical. But guarding the 3-step sequence makes the intent explicit and
future-proof:
```suggestion
kafkaProducer.beginTransaction();
try {
uow.beginTransactedBy(transactionId);
uow.addSynchronization(new
KafkaTransactionSynchronization(transactionId, kafkaProducer));
} catch (Exception uowEx) {
// beginTransaction() succeeded but UoW bookkeeping failed —
abort the open transaction
// to avoid a dangling half-begun transaction on the broker.
try {
kafkaProducer.abortTransaction();
} catch (Exception abortEx) {
uowEx.addSuppressed(abortEx);
}
throw uowEx;
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
