gnodet-bot commented on code in PR #26557:
URL: https://github.com/apache/camel/pull/26557#discussion_r4038960149
##########
components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java:
##########
@@ -521,15 +525,38 @@ 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, so a
+ // failure in beginTransaction() does not leave the unit of work
flagged as transacted without a
+ // synchronization to commit or roll it back.
kafkaProducer.beginTransaction();
- uow.addSynchronization(new
KafkaTransactionSynchronization(transactionId, kafkaProducer));
+ uow.beginTransactedBy(transactionId);
+ uow.addSynchronization(createTransactionSynchronization(exchange));
} else {
LOG.debug("Using existing kafka transaction {} with exchange {}.",
transactionId, exchange.getExchangeId());
}
}
+ private KafkaTransactionSynchronization
createTransactionSynchronization(Exchange exchange) {
+ if (configuration.isExactlyOnce()) {
+ KafkaManualCommit manual
+ =
exchange.getMessage().getHeader(KafkaConstants.MANUAL_COMMIT,
KafkaManualCommit.class);
+ if (manual instanceof DefaultKafkaManualCommit dmc) {
+ // Read the consumer group metadata on the consumer poll
thread that is processing this exchange; the
+ // Kafka consumer is not safe for multi-threaded access. The
offset to commit is the next offset to
+ // read, i.e. the processed record's offset + 1.
+ Map<TopicPartition, OffsetAndMetadata> offsets =
Collections.singletonMap(
+ dmc.getPartition(), new
OffsetAndMetadata(dmc.getRecordOffset() + 1));
Review Comment:
⚠️ **Thread-safety violation on the async code path.**
`createTransactionSynchronization(exchange)` calls
`dmc.getConsumerGroupMetadata()` (→ `consumer.groupMetadata()`) at
transaction-start time. `startKafkaTransaction` is called from **both**
`process(Exchange)` (synchronous) and `process(Exchange, AsyncCallback)`
(asynchronous, line ~458). On the async path, the producer's `process(Exchange,
AsyncCallback)` is invoked on a thread from `workerPool`, not the Kafka
consumer poll thread — violating Kafka's thread-safety contract for
`Consumer.groupMetadata()`.
The PR description acknowledges this: *"group metadata must be read on the
poll thread"*, but the code doesn't enforce it. The guard `if
(configuration.isExactlyOnce())` should reject or warn loudly when the endpoint
is configured for async delivery, or `groupMetadata()` must be read and stashed
by the consumer side before the exchange is handed off to the producer.
At minimum, add a check here:
```java
if (configuration.isExactlyOnce() && !isSynchronous()) {
throw new IllegalStateException(
"exactlyOnce requires synchronous processing (isSynchronous=true); "
+ "async delivery hands the exchange to a worker thread before the
producer runs, "
+ "making consumer.groupMetadata() unsafe to call here.");
}
```
##########
components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaTransactionSynchronization.java:
##########
@@ -46,6 +62,12 @@ public void onDone(Exchange exchange) {
kafkaProducer.abortTransaction();
}
} else {
+ // Exactly-once: commit the source consumer offsets as part of
this producer transaction so that the
+ // consumed record and the produced records are committed
atomically.
Review Comment:
⚠️ **Bug — `sendOffsetsToTransaction` failure leaves transaction open.**
If `sendOffsetsToTransaction(offsetsToCommit, groupMetadata)` throws
`KafkaException`, execution falls into the `catch (KafkaException e)` block
below, which only calls `exchange.setException(e)` — **no `abortTransaction()`
is called**. The broker-side transaction is now stuck open (fenced) until
`transaction.timeout.ms` expires. Any subsequent attempt to use this producer
will throw `ProducerFencedException`.
The `catch (Exception e)` block below it *does* call `abortTransaction()`,
but `KafkaException` is caught by the preceding block first.
Fix: abort the transaction on `sendOffsetsToTransaction` failure:
```suggestion
if (offsetsToCommit != null && groupMetadata != null) {
LOG.debug("Sending {} consumer offset(s) to kafka
transaction {}", offsetsToCommit.size(), transactionId);
try {
kafkaProducer.sendOffsetsToTransaction(offsetsToCommit, groupMetadata);
} catch (KafkaException e) {
LOG.warn("Abort kafka transaction {} due to
sendOffsetsToTransaction failure", transactionId, e);
kafkaProducer.abortTransaction();
exchange.setException(e);
return;
}
}
```
Also add a test case for this path: `sendOffsetsToTransaction` throws →
verify `abortTransaction()` is called and `commitTransaction()` is not.
##########
components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java:
##########
@@ -521,15 +525,38 @@ 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, so a
+ // failure in beginTransaction() does not leave the unit of work
flagged as transacted without a
+ // synchronization to commit or roll it back.
kafkaProducer.beginTransaction();
- uow.addSynchronization(new
KafkaTransactionSynchronization(transactionId, kafkaProducer));
+ uow.beginTransactedBy(transactionId);
+ uow.addSynchronization(createTransactionSynchronization(exchange));
} else {
LOG.debug("Using existing kafka transaction {} with exchange {}.",
transactionId, exchange.getExchangeId());
Review Comment:
💡 **Missing precondition validation for `exactlyOnce`.**
`exactlyOnce=true` without `transacted=true` or a `transactionalId` set is
silently accepted — `startKafkaTransaction` is never called (the `if
(transactionId != null)` guard skips it), and the EOS code path is never
reached. The user gets no offsets committed and no error. This should be
validated at endpoint start:
```java
if (configuration.isExactlyOnce() && transactionId == null) {
throw new IllegalArgumentException(
"exactlyOnce=true requires transacted=true or a transactionalId to
be set");
}
```
Add this to `doStart()` or `startKafkaTransaction`, whichever initialises
`transactionId`.
##########
components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConfiguration.java:
##########
@@ -2296,4 +2299,20 @@ public void setTransactionalId(String transactionalId) {
this.transactionalId = transactionalId;
}
Review Comment:
nit: `isExactlyOnce()` getter has no Javadoc. All public setters here have
one on the setter; conventionally the getter should either have its own or
point to the setter. Add at minimum:
```suggestion
/**
* Whether exactly-once (read-process-write) semantics are enabled on
this producer.
*
* @return true if exactly-once mode is active
* @see #setExactlyOnce(boolean)
*/
public boolean isExactlyOnce() {
return exactlyOnce;
}
```
--
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]
