davsclaus commented on code in PR #26557:
URL: https://github.com/apache/camel/pull/26557#discussion_r4192893549
##########
components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/DefaultKafkaManualCommit.java:
##########
@@ -64,6 +65,15 @@ public long getCommitTimeout() {
return kafkaRecordPayload.commitTimeout;
}
+ /**
+ * Gets the consumer group metadata of the source consumer, for use with
+ * {@link
org.apache.kafka.clients.producer.Producer#sendOffsetsToTransaction} when
driving exactly-once semantics.
+ * Must be called on the consumer poll thread, as the Kafka consumer is
not safe for multi-threaded access.
+ */
+ public ConsumerGroupMetadata getConsumerGroupMetadata() {
+ return camelExchangePayload.consumer.groupMetadata();
Review Comment:
This reads `groupMetadata()` lazily, on the producer's thread. With
`threads()` or `seda` between the consumer and the producer, that is not the
poll thread (`KafkaConsumer` is not thread-safe) and the generation may be
stale. A stale generation is rejected at `sendOffsetsToTransaction` as an
abortable `CommitFailedException`, so no data is corrupted, but as agreed with
Guillaume, please capture it on the poll thread (e.g. in `getManualCommit`), or
reject an exchange whose thread is not the consumer's.
##########
components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java:
##########
@@ -524,18 +533,75 @@ private void startKafkaTransaction(Exchange exchange) {
if (!uow.isTransactedBy(transactionId)) {
LOG.debug("Starting kafka transaction {} with exchange {}",
transactionId, exchange.getExchangeId());
- // 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).
+ // Resolve the synchronization first: with exactlyOnce it reads
the consumer offsets and may fail, and
+ // that must happen before the broker transaction is opened. Then
begin the broker transaction, and only
+ // then mark the unit of work. This way a failure in either step
does not leave the unit of work flagged
+ // as transacted without a synchronization to commit or roll it
back (CAMEL-24780).
+ KafkaTransactionSynchronization synchronization =
createTransactionSynchronization(exchange);
kafkaProducer.beginTransaction();
uow.beginTransactedBy(transactionId);
- uow.addSynchronization(new
KafkaTransactionSynchronization(transactionId, kafkaProducer));
+ uow.addSynchronization(synchronization);
} else {
LOG.debug("Using existing kafka transaction {} with exchange {}.",
transactionId, exchange.getExchangeId());
}
}
+ private KafkaTransactionSynchronization
createTransactionSynchronization(Exchange exchange) {
+ if (!configuration.isExactlyOnce()) {
+ return new KafkaTransactionSynchronization(transactionId,
kafkaProducer);
+ }
+
+ // Read the consumer offsets and 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.
+ List<DefaultKafkaManualCommit> manualCommits =
resolveManualCommits(exchange);
+ Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
+ for (DefaultKafkaManualCommit manualCommit : manualCommits) {
+ offsets.merge(manualCommit.getPartition(), new
OffsetAndMetadata(manualCommit.getRecordOffset() + 1),
+ (current, candidate) -> candidate.offset() >
current.offset() ? candidate : current);
+ }
+ return new KafkaTransactionSynchronization(
+ transactionId, kafkaProducer, offsets,
manualCommits.get(0).getConsumerGroupMetadata());
+ }
+
+ private List<DefaultKafkaManualCommit> resolveManualCommits(Exchange
exchange) {
+ Object body = exchange.getMessage().getBody();
+ if (body instanceof List<?> batch && !batch.isEmpty() && batch.get(0)
instanceof Exchange) {
Review Comment:
Minor: batch offsets are only collected while the body is still a
`List<Exchange>`. If the route transforms the body first, this silently falls
back to the partition of the last record, and an aggregator producing a
`List<Exchange>` also matches here (on another thread). Worth documenting both.
##########
components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java:
##########
@@ -524,18 +533,75 @@ private void startKafkaTransaction(Exchange exchange) {
if (!uow.isTransactedBy(transactionId)) {
LOG.debug("Starting kafka transaction {} with exchange {}",
transactionId, exchange.getExchangeId());
- // 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).
+ // Resolve the synchronization first: with exactlyOnce it reads
the consumer offsets and may fail, and
+ // that must happen before the broker transaction is opened. Then
begin the broker transaction, and only
+ // then mark the unit of work. This way a failure in either step
does not leave the unit of work flagged
+ // as transacted without a synchronization to commit or roll it
back (CAMEL-24780).
+ KafkaTransactionSynchronization synchronization =
createTransactionSynchronization(exchange);
kafkaProducer.beginTransaction();
uow.beginTransactedBy(transactionId);
- uow.addSynchronization(new
KafkaTransactionSynchronization(transactionId, kafkaProducer));
+ uow.addSynchronization(synchronization);
} else {
LOG.debug("Using existing kafka transaction {} with exchange {}.",
transactionId, exchange.getExchangeId());
}
}
+ private KafkaTransactionSynchronization
createTransactionSynchronization(Exchange exchange) {
+ if (!configuration.isExactlyOnce()) {
+ return new KafkaTransactionSynchronization(transactionId,
kafkaProducer);
+ }
+
+ // Read the consumer offsets and 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.
+ List<DefaultKafkaManualCommit> manualCommits =
resolveManualCommits(exchange);
+ Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
+ for (DefaultKafkaManualCommit manualCommit : manualCommits) {
+ offsets.merge(manualCommit.getPartition(), new
OffsetAndMetadata(manualCommit.getRecordOffset() + 1),
+ (current, candidate) -> candidate.offset() >
current.offset() ? candidate : current);
+ }
+ return new KafkaTransactionSynchronization(
+ transactionId, kafkaProducer, offsets,
manualCommits.get(0).getConsumerGroupMetadata());
+ }
+
+ private List<DefaultKafkaManualCommit> resolveManualCommits(Exchange
exchange) {
+ Object body = exchange.getMessage().getBody();
+ if (body instanceof List<?> batch && !batch.isEmpty() && batch.get(0)
instanceof Exchange) {
+ // batching=true: the batch exchange only carries the
manual-commit of the last record, but a poll batch
+ // can span several partitions, so every record in the batch has
to contribute its own offset. Otherwise
+ // the partitions left out of the transaction would be reprocessed
after a restart.
+ List<DefaultKafkaManualCommit> manualCommits = new
ArrayList<>(batch.size());
+ for (Object child : batch) {
+ manualCommits.add(resolveManualCommit((Exchange) child));
+ }
+ return manualCommits;
+ }
+ return List.of(resolveManualCommit(exchange));
+ }
+
+ private DefaultKafkaManualCommit resolveManualCommit(Exchange exchange) {
+ KafkaManualCommit manual =
exchange.getMessage().getHeader(KafkaConstants.MANUAL_COMMIT,
KafkaManualCommit.class);
Review Comment:
A split or multicast child also has this header (copied from the parent) and
commits its own transaction with the same offset + 1, so the output of one
source record is not atomic. See the review summary.
##########
components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaTransactionSynchronization.java:
##########
@@ -60,4 +96,24 @@ public void onDone(Exchange exchange) {
exchange.getUnitOfWork().endTransactedBy(transactionId);
}
}
+
+ /**
+ * Discards the open transaction. A fatal producer error must not be
aborted: the Kafka client contract requires the
+ * producer to be closed, and calling abortTransaction() on it only throws
again, masking the original failure.
+ */
+ private void abortOrClose(KafkaException e) {
Review Comment:
This closes the producer without rebuilding it, which is the wedge
CAMEL-24782 (#27186) fixed on main with `recover()` /
`markProducerClosedForRecreation()`. After the rebase,
`sendOffsetsToTransaction` can go inside main's `commit()` try block, so a
failure takes the same recovery path.
--
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]
