oscerd commented on code in PR #26523:
URL: https://github.com/apache/camel/pull/26523#discussion_r4029169975
##########
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:
Thanks — I'm leaving this as-is. The window is between two in-memory
operations (`beginTransactedBy` adds to a set, `addSynchronization` adds to a
list) that don't throw under normal operation; the only realistic trigger is a
JVM-level error such as `OutOfMemoryError`, in which case the follow-up
`abortTransaction()` broker call would almost certainly fail too. Wrapping it
adds nesting for a path that can't meaningfully recover, so I'd rather keep the
three-step sequence readable.
_Claude Code on behalf of @oscerd_
##########
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:
The abort-on-send-failure path is handled by the existing Camel `UnitOfWork`
lifecycle (the registered `KafkaTransactionSynchronization.onDone` →
`abortTransaction()`), which this PR does not change — it's exercised by the
transactional integration tests. Since it's pre-existing behavior rather than
something introduced here, I've kept this PR focused on the begin-ordering fix.
Happy to add dedicated `KafkaTransactionSynchronization` unit tests in a
follow-up if that's useful.
_Claude Code on behalf of @oscerd_
--
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]
