oscerd commented on code in PR #26557:
URL: https://github.com/apache/camel/pull/26557#discussion_r4044674678
##########
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:
Applied in `24b48b5`. `sendOffsetsToTransaction` is now wrapped so a
`KafkaException` aborts the transaction and sets the exception (early return),
instead of falling through to the non-aborting `catch (KafkaException)` below.
Added
`KafkaTransactionSynchronizationTest#abortsTransactionWhenSendOffsetsFails`.
Good catch, thanks!
_Claude Code on behalf of @oscerd_
##########
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:
Thanks — the underlying concern is real, but I don't think
`!isSynchronous()` is the right guard. For `from(kafka).to(kafka)` the
producer's `process(...)` (and therefore `startKafkaTransaction` /
`getConsumerGroupMetadata()`) is invoked on the consumer **poll thread**
regardless of the producer's `synchronous` option — that option selects the
send-*completion* path (the worker pool), not the thread that calls
`process()`. So a plain synchronous route is safe, and an `isSynchronous()`
check would both reject safe configs and miss the actual risk: an async handoff
*between* the consumer and the producer (e.g. `.threads()` or an async step),
where `process()` runs on another thread.
The robust fix is to capture `consumer.groupMetadata()` on the consumer side
(poll thread) into the manual-commit object and have the producer read that
snapshot, rather than reading it lazily. I've flagged that as the intended
follow-up and documented the current limitation in the PR description. Keeping
it as an open design point for this draft rather than adding the
`isSynchronous` guard — does capturing on the consumer side sound right to you?
_Claude Code on behalf of @oscerd_
##########
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:
Applied in `24b48b5` — `doStart()` now throws `IllegalArgumentException`
when `exactlyOnce=true` without a transaction (`transacted=true` or a
`transactionalId`).
_Claude Code on behalf of @oscerd_
##########
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:
Applied in `24b48b5`.
_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]
