gnodet-bot commented on code in PR #28730:
URL: https://github.com/apache/camel/pull/28730#discussion_r4432627728
##########
components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/streaming/KafkaRecordStreamingProcessor.java:
##########
@@ -104,6 +117,107 @@ 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);
Review Comment:
⚠️ **Filtered records replay forever when any route branch contains a
wireTap, recipientList, or routingSlip**
`containsAsynchronousNode(route.getRoute())` walks the **entire route
tree**, including branches the exchange never took. For example:
```java
from(eosSource).filter(predicate).to(eosSink).end().wireTap("direct:audit");
```
A filtered record (bypasses `.to(eosSink)`) reaches
`completeExactlyOnceProcessing` with `!metadata.isClaimed()`.
`hasAsynchronousHandoff` walks the whole route → finds `wireTap` → returns
`true` → sets an exception → consumer replays the record forever.
With this commit, `recipientList` and `routingSlip` are now also
unconditionally flagged as async. Any route containing those EIPs — regardless
of whether the filtered record ever reached them — causes filtered records to
replay forever. This is a regression for the `filter + recipientList` pattern.
The `TO_ENDPOINT` check at lines 184–186 is already the right approach (it
reflects what the exchange actually did). The static route-tree walk is not.
Fix options:
1. Remove the tree walk entirely — rely on `TO_ENDPOINT` for `seda:`
detection and add a flag that async dispatchers (wireTap, threads) set on the
exchange when they actually fire.
2. Only walk the route tree when `TO_ENDPOINT` is not set and limit
detection to direct children of the route root (not deep children in branches).
Also: `org.apache.camel.Route` is fully qualified here; please add `import
org.apache.camel.Route;` and use the simple name.
##########
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);
Review Comment:
⚠️ **Record loss: DLQ validation rejection counted as a Kafka send failure**
`validateExactlyOnceDestination` records its own rejection as a Kafka send
failure via `recordSendFailure`. This enables a silent record-loss path:
1. Processing fails (non-Kafka exception, e.g. an `IllegalStateException`
from a `process()` step).
2. DLC routes the exchange to `kafka:dlq` (a plain Kafka endpoint, no
`exactlyOnce=true`).
3. `validateExactlyOnceDestination` fires:
`isSecondaryKafkaDeadLetterSend()` returns false (no prior Kafka send failure),
so it calls `recordSendFailure` and throws `IllegalStateException`.
4. DLC handles its own new exception (`deadLetterHandleNewException=true`) —
`exchange.getException()` is cleared.
5. `KafkaTransactionSynchronization.onDone()`: `sendFailure != null`, no
exception → takes `commitOffsetsAfterHandledSendFailure` → aborts the
(successfully sent) `eosSink` output and commits only source offsets. DLQ topic
is empty.
The record is silently lost. The offset-skip path was designed for a genuine
Kafka send failure handled by the error handler. A DLQ configuration rejection
is not a Kafka send failure. Fix: remove the `recordSendFailure` call from
`validateExactlyOnceDestination` and instead mark the `exactlyOnceMetadata`
claim as rejected (analogous to the existing `rejectClaim` pattern) so `onDone`
takes the rollback path rather than the offset-skip 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]
