This is an automated email from the ASF dual-hosted git repository.
gnodet pushed a commit to branch camel-4.22.x
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/camel-4.22.x by this push:
new 6c160c610450 [backport camel-4.22.x] CAMEL-24780: camel-kafka - fix
transactional producer leaking a half-begun transaction and skipping the async
callback (#26555)
6c160c610450 is described below
commit 6c160c610450275b96c7019857afd70565ae5d82
Author: Guillaume Nodet <[email protected]>
AuthorDate: Fri Sep 18 14:31:54 2026 +0200
[backport camel-4.22.x] CAMEL-24780: camel-kafka - fix transactional
producer leaking a half-begun transaction and skipping the async callback
(#26555)
---
.../camel/component/kafka/KafkaProducer.java | 18 ++++++++----
.../camel/component/kafka/KafkaProducerTest.java | 32 ++++++++++++++++++++++
2 files changed, 44 insertions(+), 6 deletions(-)
diff --git
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
index 7c8c7d722e66..55e57f6e623a 100755
---
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
+++
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
@@ -476,13 +476,16 @@ public class KafkaProducer extends DefaultAsyncProducer
implements RouteIdAware
= new KafkaProducerCallBack(exchange, callback, workerPool,
configuration.isRecordMetadata());
Message message = exchange.getMessage();
- Object body = message.getBody();
-
- if (transactionId != null) {
- startKafkaTransaction(exchange);
- }
try {
+ Object body = message.getBody();
+
+ // Start the transaction inside the try so that a failure to begin
it (or a failing lazy body
+ // conversion) still completes the async callback instead of
escaping process() (CAMEL-24780).
+ if (transactionId != null) {
+ startKafkaTransaction(exchange);
+ }
+
// is the message body a list or something that contains multiple
values
if (endpoint.getConfiguration().isUseIterator() &&
isIterable(body)) {
processIterableAsync(exchange, producerCallBack, message);
@@ -542,8 +545,11 @@ public class KafkaProducer extends DefaultAsyncProducer
implements RouteIdAware
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));
} else {
LOG.debug("Using existing kafka transaction {} with exchange {}.",
diff --git
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerTest.java
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerTest.java
index 67db580eb5f4..b195eae3235a 100755
---
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerTest.java
+++
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerTest.java
@@ -42,6 +42,7 @@ import org.apache.camel.impl.DefaultCamelContext;
import org.apache.camel.impl.engine.DefaultHeadersMapFactory;
import org.apache.camel.processor.aggregate.GroupedExchangeAggregationStrategy;
import org.apache.camel.processor.aggregate.GroupedMessageAggregationStrategy;
+import org.apache.camel.spi.UnitOfWork;
import org.apache.camel.support.DefaultExchange;
import org.apache.camel.support.DefaultMessage;
import org.apache.kafka.clients.producer.Callback;
@@ -59,6 +60,7 @@ import static
org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.ArgumentMatchers.isA;
@@ -189,6 +191,36 @@ public class KafkaProducerTest {
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.
+ // Exercise the real init path: doStart() derives transactionId from
transactionalId and calls
+ // initTransactions() on the mock (a no-op), so no reflection on the
private field is needed.
+ endpoint.getConfiguration().setTransactionalId("test-tx");
+ endpoint.getConfiguration().setTopic("sometopic");
+ producer.doStart();
+
+ 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());
+ }
+
@Test
public void processSendsMessageWithTopicHeaderAndNoTopicInEndPoint()
throws Exception {
endpoint.getConfiguration().setTopic(null);
