gnodet-bot commented on code in PR #28730:
URL: https://github.com/apache/camel/pull/28730#discussion_r4432313767
##########
components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/streaming/KafkaRecordStreamingProcessor.java:
##########
@@ -104,6 +117,102 @@ public ProcessingResult processExchange(
return result;
}
+ /**
+ * A route that did not claim the source metadata with an exactly-once
producer still has to advance the source
+ * offset (for example, a filter branch). Run this after route completion
on the consumer polling thread.
+ */
+ private void completeExactlyOnceProcessing(Exchange exchange) {
+ if (!configuration.isExactlyOnce()) {
+ return;
+ }
+
+ KafkaExactlyOnceMetadata metadata = exchange.getProperty(
+ KafkaExactlyOnceMetadata.EXCHANGE_PROPERTY,
KafkaExactlyOnceMetadata.class);
+ if (metadata != null && metadata.isClaimRejected()) {
+ if (exchange.getException() == null) {
+ exchange.setException(new IllegalStateException(
+ "The Kafka exactlyOnce source metadata was rejected by
another exchange or producer: "
+ +
metadata.getClaimRejectionReason()));
+ }
+ return;
+ }
+
+ boolean transactionAborted = Boolean.TRUE.equals(
+
exchange.getProperty(KafkaExactlyOnceMetadata.TRANSACTION_ABORTED_PROPERTY));
+ if (transactionAborted || exchange.isRollbackOnly()) {
+ if (exchange.getException() == null) {
+ exchange.setException(new IllegalStateException(
+ "The Kafka exactlyOnce transaction was rolled back;
the source record must be replayed"));
+ }
+ return;
+ }
+ if (exchange.getException() != null) {
+ return;
+ }
+
+ Throwable sendFailure = metadata != null ? metadata.getSendFailure() :
null;
+ if (sendFailure != null && !metadata.isClaimed()) {
+ // A producer failed before it could register a transaction
synchronization. Do not mistake a handled
+ // configuration/startup failure for a filter branch and commit
its source offset.
+ exchange.setException(new IllegalStateException(
+ "The exactlyOnce Kafka producer failed before it could
claim the source offsets", sendFailure));
+ return;
+ }
+ if (metadata == null) {
+ exchange.setException(new IllegalStateException(
+ "exactlyOnce source metadata was not captured for the
consumed record"));
+ return;
+ }
+ if (!metadata.isClaimed()) {
+ if (hasAsynchronousHandoff(exchange)) {
+ exchange.setException(new IllegalStateException(
+ "An exactlyOnce Kafka source cannot commit offsets
directly when the route hands the exchange "
+ + "to an
asynchronous endpoint before an exactly-once Kafka producer claims it; "
+ + "invoke the
destination synchronously from the source exchange"));
+ return;
+ }
+ try {
+ metadata.commitOffsets();
+ } catch (Exception commitFailure) {
+ exchange.setException(commitFailure);
+ }
+ }
+ }
+
+ /** Detect asynchronous route constructs when no synchronous Kafka
producer claimed the source offsets. */
+ private boolean hasAsynchronousHandoff(Exchange exchange) {
+ String toEndpoint =
exchange.getProperty(ExchangePropertyKey.TO_ENDPOINT, String.class);
+ if (toEndpoint != null && toEndpoint.startsWith("seda:")) {
+ return true;
+ }
+ String routeId = exchange.getFromRouteId();
+ if (routeId == null) {
+ return false;
+ }
+ org.apache.camel.Route route = exchange.getContext().getRoute(routeId);
+ return route != null && containsAsynchronousNode(route.getRoute());
+ }
+
+ private boolean containsAsynchronousNode(NamedNode node) {
+ if (node == null) {
+ return false;
+ }
+ String shortName = node.getShortName();
+ if ("wireTap".equals(shortName) || "threads".equals(shortName)) {
+ return true;
+ }
+ String label = node.getLabel();
+ if ("to".equals(shortName) && label != null &&
label.startsWith("to[seda:")) {
+ return true;
+ }
+ for (NamedNode child : node.getChildren()) {
+ if (containsAsynchronousNode(child)) {
+ return true;
+ }
+ }
+ return false;
+ }
Review Comment:
💡 **Gap: `toD`, `recipientList`, and `routingSlip` are not detected as async
handoffs**
`containsAsynchronousNode()` checks for `"wireTap"`, `"threads"`, and static
`"to[seda:...]"`, but misses:
- `toD("seda:...")` — shortName is `"toD"`, not `"to"`, so line 216 never
matches
- `recipientList` / `routingSlip` — can dynamically route to a `seda:`
endpoint at runtime
A route using `toD("seda:my-queue")` would pass this check undetected.
Consider either documenting the known gap (a `@implNote` or TODO), or expanding
detection to cover `"toD"` with the same label check.
##########
components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java:
##########
@@ -568,29 +600,216 @@ private void startKafkaTransaction(Exchange exchange) {
UnitOfWork uow = exchange.getUnitOfWork();
if (!uow.isTransactedBy(transactionId)) {
- // A prior transaction may have hit a fatal error that closed the
shared producer; rebuild it before
- // starting a new transaction so the route recovers instead of
staying wedged on a dead producer
- // (CAMEL-24782).
- recreateProducerIfClosed();
- 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).
- kafkaProducer.beginTransaction();
- uow.beginTransactedBy(transactionId);
- uow.addSynchronization(new
KafkaTransactionSynchronization(transactionId, kafkaProducer, this));
+ // Resolve and validate source metadata before opening the broker
transaction. A failure here must not
+ // leave an open transaction without a registered synchronization.
+ KafkaExactlyOnceMetadata exactlyOnceMetadata
+ = configuration.isExactlyOnce() ?
createExactlyOnceMetadata(exchange) : null;
+ if (exactlyOnceMetadata != null) {
+ String registeredProducer = exchange.getProperty(
+
KafkaConstants.EXACTLY_ONCE_PRODUCER_TRANSACTION_ID_PROPERTY, String.class);
+ if (registeredProducer != null &&
!registeredProducer.equals(transactionId)) {
+ String reason = "Only one exactlyOnce Kafka producer is
supported per exchange/unit of work";
+ exactlyOnceMetadata.rejectClaim(reason);
+ throw new
KafkaExactlyOnceMetadata.ClaimRejectedException(reason);
+ }
+ if (!activeExactlyOnceExchange.compareAndSet(null,
exchange.getExchangeId())) {
+ throw new KafkaExactlyOnceMetadata.ClaimRejectedException(
+ "A Kafka exactlyOnce transaction is already active
for exchange "
+
+ activeExactlyOnceExchange.get()
+
+ "; concurrent exchanges cannot share this transactional "
+
+ "producer. Serialize processing before the destination producer.");
+ }
+ }
+
+ boolean transactionStarted = false;
+ boolean unitOfWorkMarked = false;
+ boolean synchronizationRegistered = false;
+ boolean metadataClaimed = false;
+ try {
+ if (exactlyOnceMetadata != null) {
+ // The metadata object is shared by shallow exchange
copies. Bind it to the original source exchange
+ // and allow only one producer transaction to claim its
offsets.
+ exactlyOnceMetadata.claim(exchange.getExchangeId());
+ metadataClaimed = true;
+ }
+ // A prior transaction may have hit a fatal error that closed
the shared producer; rebuild it before
+ // starting a new transaction so the route recovers instead of
staying wedged on a dead producer
+ // (CAMEL-24782).
+ recreateProducerIfClosed();
+ 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).
+ kafkaProducer.beginTransaction();
+ transactionStarted = true;
+ uow.beginTransactedBy(transactionId);
+ unitOfWorkMarked = true;
+ uow.addSynchronization(new KafkaTransactionSynchronization(
+ transactionId, kafkaProducer, this,
exactlyOnceMetadata, exchange.getExchangeId()));
+ synchronizationRegistered = true;
+ if (exactlyOnceMetadata != null) {
+ // A marker from an earlier attempt that failed before
synchronization registration does not
+ // belong to this transaction. Start the newly registered
attempt with a clean send-failure state.
+ exactlyOnceMetadata.clearSendFailures();
+
exchange.removeProperty(KafkaExactlyOnceMetadata.SEND_FAILURE_PROPERTY);
+
exchange.removeProperty(KafkaExactlyOnceMetadata.SECONDARY_SEND_FAILURE_PROPERTY);
+
exchange.setProperty(KafkaConstants.EXACTLY_ONCE_PRODUCER_TRANSACTION_ID_PROPERTY,
transactionId);
+ }
+ } catch (RuntimeException | Error startFailure) {
+ // If no synchronization owns the transaction yet, clean up
here. Once registered, onDone()
+ // owns both transaction completion and release of the
exactly-once producer slot.
+ if (!synchronizationRegistered) {
+ if (metadataClaimed) {
+ exactlyOnceMetadata.releaseClaim();
+ }
+ if (unitOfWorkMarked) {
+ try {
+ uow.endTransactedBy(transactionId);
+ } catch (RuntimeException cleanupFailure) {
+ startFailure.addSuppressed(cleanupFailure);
+ }
+ }
+ if (transactionStarted) {
+ try {
+ kafkaProducer.abortTransaction();
+ } catch (Exception abortFailure) {
+ startFailure.addSuppressed(abortFailure);
+ try {
+ kafkaProducer.close();
+ } catch (Exception closeFailure) {
+ startFailure.addSuppressed(closeFailure);
+ } finally {
+ markProducerClosedForRecreation();
+ }
+ }
+ } else if
(KafkaTransactionSynchronization.isFatal(startFailure)) {
+ // beginTransaction() can itself fence or otherwise
fatally invalidate the producer.
+ try {
+ kafkaProducer.close();
+ } catch (Exception closeFailure) {
+ startFailure.addSuppressed(closeFailure);
+ } finally {
+ markProducerClosedForRecreation();
+ }
+ }
+ if (exactlyOnceMetadata != null) {
+
releaseExactlyOnceTransaction(exchange.getExchangeId());
+ }
+ }
+ throw startFailure;
+ }
} else {
+ if (configuration.isExactlyOnce()) {
+ Throwable previousSendFailure = exchange.getProperty(
+ KafkaExactlyOnceMetadata.SEND_FAILURE_PROPERTY,
Throwable.class);
+ if (previousSendFailure != null) {
+ // Do not restart the transaction here: Camel redelivers
from the failed processor, so output
+ // records sent by earlier route steps in this transaction
would be discarded without replay.
+ // Roll back the whole source exchange and let the Kafka
consumer replay it in a new transaction.
+ // This synthetic rejection is not a second Kafka send
failure. Preserve only the original
+ // failure so a successfully handled non-Kafka DLQ can
still take the offset-only skip path.
+
exchange.setProperty(KafkaExactlyOnceMetadata.SUPPRESS_NEXT_SEND_FAILURE_PROPERTY,
Boolean.TRUE);
+ exchange.setRollbackOnly(true);
+ throw new IllegalStateException(
+ "Cannot retry a Kafka exactlyOnce send within an
active transaction; "
+ + "the transaction is
marked for rollback and the source record "
+ + "must be replayed in a
new transaction",
+ previousSendFailure);
+ }
+ }
LOG.debug("Using existing kafka transaction {} with exchange {}.",
transactionId, exchange.getExchangeId());
}
}
+ private void validateExactlyOnceDestination(Exchange exchange) {
+ if (exchange.getFromEndpoint() instanceof KafkaEndpoint sourceEndpoint
+ && sourceEndpoint.getConfiguration().isExactlyOnce()
+ && (!configuration.isExactlyOnce() || transactionId == null)
+ && !isSecondaryKafkaDeadLetterSend(exchange)) {
+ IllegalStateException failure = new IllegalStateException(
+ "A Kafka exactlyOnce source requires every Kafka
destination producer to set exactlyOnce=true "
+ + "and
configure an explicit transactionalId");
+ KafkaTransactionSynchronization.recordSendFailure(exchange,
failure);
+ throw failure;
+ }
+ }
+
+ /** A Kafka dead-letter producer is allowed only after the source producer
has recorded a send failure. */
+ private boolean isSecondaryKafkaDeadLetterSend(Exchange exchange) {
+ KafkaExactlyOnceMetadata metadata = exchange.getProperty(
+ KafkaExactlyOnceMetadata.EXCHANGE_PROPERTY,
KafkaExactlyOnceMetadata.class);
+ String sourceProducer = exchange.getProperty(
+ KafkaConstants.EXACTLY_ONCE_PRODUCER_TRANSACTION_ID_PROPERTY,
String.class);
+ return metadata != null && metadata.getSendFailure() != null &&
sourceProducer != null
+ && !Objects.equals(sourceProducer, transactionId);
+ }
+
+ private KafkaExactlyOnceMetadata createExactlyOnceMetadata(Exchange
exchange) {
+ KafkaExactlyOnceMetadata metadata
+ =
exchange.getProperty(KafkaExactlyOnceMetadata.EXCHANGE_PROPERTY,
KafkaExactlyOnceMetadata.class);
+
+ // The original source producer already owns the offsets and its
transaction has failed. An exactly-once
+ // Kafka DLQ producer may now commit its DLQ record in its own
transaction; the source synchronization is
+ // registered first and therefore commits offsets only after this
later synchronization has completed.
+ if (isSecondaryKafkaDeadLetterSend(exchange)) {
+ return null;
+ }
+
+ if (!(exchange.getFromEndpoint() instanceof KafkaEndpoint
sourceEndpoint)) {
+ if (metadata != null) {
+ metadata.rejectCopyClaim(
+ "exactlyOnce source metadata cannot be claimed from a
non-Kafka route or copied exchange");
+ }
+ throw new KafkaExactlyOnceMetadata.ClaimRejectedException(
+ "exactlyOnce requires a Kafka consumer as the route
source");
+ }
+
+ KafkaConfiguration source = sourceEndpoint.getConfiguration();
+ if (!source.isExactlyOnce() || !source.isAllowManualCommit()
+ || source.isAutoCommitEnable() ||
!source.isBreakOnFirstError()) {
+ if (metadata != null) {
+ metadata.rejectCopyClaim("exactlyOnce source endpoint
configuration is incompatible with source metadata");
+ }
+ throw new KafkaExactlyOnceMetadata.ClaimRejectedException(
+ "exactlyOnce requires the Kafka source endpoint to set
exactlyOnce=true, "
+ +
"allowManualCommit=true, autoCommitEnable=false and breakOnFirstError=true");
+ }
+
+ if (metadata == null) {
+ KafkaManualCommit manualCommit
+ =
exchange.getMessage().getHeader(KafkaConstants.MANUAL_COMMIT,
KafkaManualCommit.class);
+ if (manualCommit == null) {
+ throw new IllegalStateException(
+ "exactlyOnce requires source offset metadata;
configure allowManualCommit=true "
+ + "on the Kafka source
endpoint");
+ }
+ String sourceExchangeId = null;
+ if (manualCommit instanceof DefaultKafkaManualCommit defaultCommit
+ && defaultCommit.getCamelExchangePayload() != null
+ && defaultCommit.getCamelExchangePayload().exchange !=
null) {
+ sourceExchangeId =
defaultCommit.getCamelExchangePayload().exchange.getExchangeId();
+ }
+ metadata = KafkaExactlyOnceMetadata.fromManualCommits(
+ Collections.singletonList(manualCommit), sourceExchangeId);
+ exchange.setProperty(KafkaExactlyOnceMetadata.EXCHANGE_PROPERTY,
metadata);
+ }
+
+ if (ObjectHelper.isNotEmpty(source.getGroupId())
+ && !Objects.equals(source.getGroupId(),
metadata.getConsumerGroupMetadata().groupId())) {
+ String reason = "exactlyOnce source groupId does not match the
consumer group metadata captured for the record";
+ metadata.rejectClaim(reason);
+ throw new IllegalStateException(reason);
Review Comment:
⚠️ **Bug: wrong exception type for `groupId` mismatch**
Every other rejection path in `createExactlyOnceMetadata()` throws
`ClaimRejectedException` (lines 764 and 774), but this path throws plain
`IllegalStateException`. The consequence is silent misclassification:
`recordSendFailure()` only filters out `ClaimRejectedException` — a plain
`IllegalStateException` slips through and gets recorded as a Kafka send
failure. `KafkaTransactionSynchronization.onDone()` then takes the offset-only
commit path instead of rolling back and replaying.
```suggestion
throw new
KafkaExactlyOnceMetadata.ClaimRejectedException(reason);
```
--
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]
