gnodet-bot commented on code in PR #28730:
URL: https://github.com/apache/camel/pull/28730#discussion_r4445582318
##########
components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/streaming/KafkaRecordStreamingProcessor.java:
##########
@@ -104,6 +162,517 @@ 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.isDeadLetterSendFailed()) {
+ if (exchange.getException() == null) {
+ exchange.setException(new IllegalStateException(
+ "A Kafka dead-letter send failed; the source record
must be replayed"));
+ }
+ return;
+ }
+ 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 hand-offs whose target route can reach Kafka. The
event notifier marks the shared metadata
+ * object, so the original source observes a Kafka-directed hand-off even
when Camel emits the event for a copied
+ * exchange. A statically non-Kafka asynchronous target must not affect
the source exchange.
+ */
+ boolean hasAsynchronousHandoff(Exchange exchange) {
+ String toEndpoint =
exchange.getProperty(ExchangePropertyKey.TO_ENDPOINT, String.class);
+ KafkaExactlyOnceMetadata metadata = exchange.getProperty(
+ KafkaExactlyOnceMetadata.EXCHANGE_PROPERTY,
KafkaExactlyOnceMetadata.class);
+ return asynchronousEndpointCanReachKafka(exchange.getContext(),
toEndpoint)
+ || metadata != null && metadata.hasAsynchronousKafkaHandoff();
+ }
+
+ /**
+ * Register an event notifier while at least one exactly-once Kafka source
consumer is active in this context.
+ * Events without exactly-once metadata return before URI inspection,
avoiding work for unrelated routes.
+ */
+ static void registerAsynchronousHandoffTracking(CamelContext context) {
+ if (context == null) {
+ return;
+ }
+ synchronized (ASYNCHRONOUS_HANDOFF_TRACKED_CONTEXTS) {
+ HandoffTrackingRegistration existing =
ASYNCHRONOUS_HANDOFF_TRACKED_CONTEXTS.get(context);
+ if (existing != null) {
+ existing.references++;
+ return;
+ }
+
+ ConcurrentMap<String, Boolean> asynchronousKafkaTargets = new
ConcurrentHashMap<>();
+ EventNotifierSupport notifier = new EventNotifierSupport() {
+ @Override
+ public void notify(CamelEvent event) {
+ if (event instanceof CamelEvent.ExchangeSendingEvent
sendingEvent) {
+ Exchange exchange = sendingEvent.getExchange();
+ KafkaExactlyOnceMetadata metadata = metadata(exchange);
+ if (metadata == null) {
+ return;
+ }
+ String uri =
sendingEvent.getEndpoint().getEndpointUri();
+ if (asynchronousEndpointCanReachKafka(context, uri,
asynchronousKafkaTargets)) {
+ // The metadata object is shared by shallow
exchange copies. Mark it only when this
+ // asynchronous target can reach Kafka; a
non-Kafka WireTap must not poison its source.
+ metadata.markAsynchronousKafkaHandoff();
+ }
+ return;
+ }
+
+ if (event instanceof
CamelEvent.ExchangeFailureHandlingEvent handlingEvent) {
+ Exchange exchange = handlingEvent.getExchange();
+ if (metadata(exchange) == null) {
+ return;
+ }
+ Object current = exchange.getProperty(
+
KafkaExactlyOnceMetadata.FAILURE_HANDLER_DELIVERY_DEPTH_PROPERTY);
+ int depth = current instanceof Integer value ? value :
0;
+ exchange.setProperty(
+
KafkaExactlyOnceMetadata.FAILURE_HANDLER_DELIVERY_DEPTH_PROPERTY, depth + 1);
+ exchange.setProperty(
+
KafkaExactlyOnceMetadata.FAILURE_HANDLER_DELIVERY_PROPERTY, Boolean.TRUE);
+ return;
+ }
+
+ if (event instanceof
CamelEvent.ExchangeFailureHandledEvent handledEvent) {
+ Exchange exchange = handledEvent.getExchange();
+ if (metadata(exchange) == null) {
+ return;
+ }
+ Object current = exchange.getProperty(
+
KafkaExactlyOnceMetadata.FAILURE_HANDLER_DELIVERY_DEPTH_PROPERTY);
+ int depth = current instanceof Integer value ? value :
0;
+ if (depth <= 1) {
+ exchange.removeProperty(
+
KafkaExactlyOnceMetadata.FAILURE_HANDLER_DELIVERY_DEPTH_PROPERTY);
+ exchange.removeProperty(
+
KafkaExactlyOnceMetadata.FAILURE_HANDLER_DELIVERY_PROPERTY);
+ } else {
+ exchange.setProperty(
+
KafkaExactlyOnceMetadata.FAILURE_HANDLER_DELIVERY_DEPTH_PROPERTY, depth - 1);
+ }
+ }
+ }
+ };
+ notifier.setIgnoreCamelContextEvents(true);
+ notifier.setIgnoreCamelContextInitEvents(true);
+ notifier.setIgnoreRouteEvents(true);
+ notifier.setIgnoreServiceEvents(true);
+ notifier.setIgnoreStepEvents(true);
+ notifier.setIgnoreExchangeCreatedEvent(true);
+ notifier.setIgnoreExchangeCompletedEvent(true);
+ // FailureHandling/FailureHandled are ExchangeFailureEvent
variants. Their scoped window covers the
+ // selected failure processor (including
onException(...).handled(true)), not sends after doCatch or
+ // onException(...).continued(true).
+ notifier.setIgnoreExchangeFailedEvents(false);
+ notifier.setIgnoreExchangeRedeliveryEvents(true);
+ notifier.setIgnoreExchangeSentEvents(true);
+ notifier.setIgnoreExchangeSendingEvents(false);
+ context.getManagementStrategy().addEventNotifier(notifier);
+ ASYNCHRONOUS_HANDOFF_TRACKED_CONTEXTS.put(
+ context, new HandoffTrackingRegistration(notifier,
asynchronousKafkaTargets));
+ }
+ }
+
+ /** Remove the notifier when the final exactly-once Kafka source consumer
stops. */
+ static void unregisterAsynchronousHandoffTracking(CamelContext context) {
+ if (context == null) {
+ return;
+ }
+ synchronized (ASYNCHRONOUS_HANDOFF_TRACKED_CONTEXTS) {
+ HandoffTrackingRegistration registration =
ASYNCHRONOUS_HANDOFF_TRACKED_CONTEXTS.get(context);
+ if (registration == null) {
+ return;
+ }
+ if (--registration.references == 0) {
+
context.getManagementStrategy().removeEventNotifier(registration.notifier);
+ ASYNCHRONOUS_HANDOFF_TRACKED_CONTEXTS.remove(context);
+ }
+ }
+ }
+
+ private static boolean asynchronousEndpointCanReachKafka(CamelContext
context, String uri) {
+ if (!isAsynchronousHandoffEndpoint(uri)) {
+ return false;
+ }
+ if (context == null) {
+ // Without a context the target graph cannot be inspected, so fail
closed.
+ return true;
+ }
+
+ HandoffTrackingRegistration registration;
+ synchronized (ASYNCHRONOUS_HANDOFF_TRACKED_CONTEXTS) {
+ registration = ASYNCHRONOUS_HANDOFF_TRACKED_CONTEXTS.get(context);
+ }
+ if (registration == null) {
+ return routesToKafka(context, uri, new HashSet<>());
+ }
+ return asynchronousEndpointCanReachKafka(context, uri,
registration.asynchronousKafkaTargets);
+ }
+
+ private static boolean asynchronousEndpointCanReachKafka(
+ CamelContext context, String uri, ConcurrentMap<String, Boolean>
cache) {
+ if (!isAsynchronousHandoffEndpoint(uri)) {
+ return false;
+ }
+ if (context == null) {
+ return true;
+ }
+ String target = endpointKey(uri);
+ return cache.computeIfAbsent(target, key -> routesToKafka(context,
uri, new HashSet<>()));
Review Comment:
⚠️ **Stale async-target cache: `false` entries are never evicted on dynamic
route changes**
`computeIfAbsent` populates the cache the first time a given
SEDA/vm/disruptor URI is seen. If a user adds a route at runtime that makes a
previously non-Kafka target now route to Kafka, the cached `false` is never
evicted. On the next exactly-once exchange, `hasAsynchronousHandoff` returns
`false` for the hand-off endpoint and the source commits its offset without the
Kafka send — silently violating exactly-once.
The cache is held in `HandoffTrackingRegistration`, which lives as long as
any exactly-once consumer is active. Routes added during that window (e.g.
dynamically via `CamelContext.addRoutes()` or by a route template) will not
invalidate it.
Mitigations (pick one):
1. **Listen for `RouteAddedEvent`** and clear the affected cache entry when
a new route's input URI matches a cached key.
2. **Document the limitation explicitly** (cache is valid only for the route
topology at consumer start; dynamic route additions after that are not tracked).
3. **Skip the cache for the no-registration path** — the uncached
`routesToKafka(context, uri, new HashSet<>())` call is the safe conservative
fallback; the cache is only worth it for the hot event-notifier path where
throughput matters.
If option 2, add a note in the Javadoc of
`asynchronousEndpointCanReachKafka(CamelContext, String, ConcurrentMap)` so
future maintainers understand the invariant.
##########
components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/streaming/KafkaRecordStreamingProcessor.java:
##########
@@ -104,6 +162,517 @@ 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.isDeadLetterSendFailed()) {
+ if (exchange.getException() == null) {
+ exchange.setException(new IllegalStateException(
+ "A Kafka dead-letter send failed; the source record
must be replayed"));
+ }
+ return;
+ }
+ 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 hand-offs whose target route can reach Kafka. The
event notifier marks the shared metadata
+ * object, so the original source observes a Kafka-directed hand-off even
when Camel emits the event for a copied
+ * exchange. A statically non-Kafka asynchronous target must not affect
the source exchange.
+ */
+ boolean hasAsynchronousHandoff(Exchange exchange) {
+ String toEndpoint =
exchange.getProperty(ExchangePropertyKey.TO_ENDPOINT, String.class);
+ KafkaExactlyOnceMetadata metadata = exchange.getProperty(
+ KafkaExactlyOnceMetadata.EXCHANGE_PROPERTY,
KafkaExactlyOnceMetadata.class);
+ return asynchronousEndpointCanReachKafka(exchange.getContext(),
toEndpoint)
+ || metadata != null && metadata.hasAsynchronousKafkaHandoff();
+ }
+
+ /**
+ * Register an event notifier while at least one exactly-once Kafka source
consumer is active in this context.
+ * Events without exactly-once metadata return before URI inspection,
avoiding work for unrelated routes.
+ */
+ static void registerAsynchronousHandoffTracking(CamelContext context) {
+ if (context == null) {
+ return;
+ }
+ synchronized (ASYNCHRONOUS_HANDOFF_TRACKED_CONTEXTS) {
+ HandoffTrackingRegistration existing =
ASYNCHRONOUS_HANDOFF_TRACKED_CONTEXTS.get(context);
+ if (existing != null) {
+ existing.references++;
+ return;
+ }
+
+ ConcurrentMap<String, Boolean> asynchronousKafkaTargets = new
ConcurrentHashMap<>();
+ EventNotifierSupport notifier = new EventNotifierSupport() {
+ @Override
+ public void notify(CamelEvent event) {
+ if (event instanceof CamelEvent.ExchangeSendingEvent
sendingEvent) {
+ Exchange exchange = sendingEvent.getExchange();
+ KafkaExactlyOnceMetadata metadata = metadata(exchange);
+ if (metadata == null) {
+ return;
+ }
+ String uri =
sendingEvent.getEndpoint().getEndpointUri();
+ if (asynchronousEndpointCanReachKafka(context, uri,
asynchronousKafkaTargets)) {
+ // The metadata object is shared by shallow
exchange copies. Mark it only when this
+ // asynchronous target can reach Kafka; a
non-Kafka WireTap must not poison its source.
+ metadata.markAsynchronousKafkaHandoff();
+ }
+ return;
+ }
+
+ if (event instanceof
CamelEvent.ExchangeFailureHandlingEvent handlingEvent) {
+ Exchange exchange = handlingEvent.getExchange();
+ if (metadata(exchange) == null) {
Review Comment:
⚠️ **`doCatch` sends are now classified as failure-handler deliveries**
Removing the `isDeadLetterChannel()` guard expands the event window to
include `CatchProcessor`-fired events. Camel's `CatchProcessor` calls:
```java
EventHelper.notifyExchangeFailureHandling(exchange.getContext(), exchange,
processor, false, null);
// ... catch body executes ...
EventHelper.notifyExchangeFailureHandled(exchange.getContext(), exchange,
processor, false, null);
```
with `deadLetterChannel=false`. Under the new code,
`FAILURE_HANDLER_DELIVERY_PROPERTY` is set to `true` for the entire duration of
the `doCatch` body. A Kafka send inside a `doCatch` block will therefore be
treated as a failure-handler delivery — not as a normal route send.
The documentation explicitly states the opposite: *"Sends after a `doCatch`
block... are normal route sends, not failure-handler deliveries."* This is
consistent for sends **after** the `doCatch` block ends (since
`ExchangeFailureHandledEvent` clears the property), but inconsistent for sends
**inside** the `doCatch` body.
Concrete impact: with exactly-once mode, a route like:
```java
from(eosSource)
.doTry().to(eosSink)
.doCatch(Exception.class).to("kafka:dlq") // <-- now classified as
failure-handler delivery
.end();
```
will send the `kafka:dlq` record on the failure-handler delivery path rather
than the normal path.
Please either:
1. **Add the `isDeadLetterChannel()` check back for the `doCatch` path** —
and separately handle `onException(...).handled(true)` via a different
detection mechanism (the `isDeadLetterChannel()` flag is `false` for both; the
difference is the processor type: `TryProcessor` vs `RedeliveryErrorHandler`).
Alternatively, inspect the failure processor type in the event.
2. **Document that sends inside `doCatch` are treated as failure-handler
deliveries**, and update the doc from "Sends after a `doCatch` block are normal
sends" to "Sends inside or after a `doCatch` block are treated as
failure-handler deliveries if they reach a Kafka endpoint."
3. **Add an IT** that sends to a Kafka endpoint inside a `doCatch` body and
asserts the expected transaction behavior, so the behavior is at least pinned
and visible.
##########
components/camel-kafka/src/main/docs/kafka-component.adoc:
##########
@@ -552,10 +555,157 @@ At the end of exchange routing, the kafka producer would
commit the transaction
It would work with JTA `camel-jta` by using `transacted()` and if it involves
some resources (SQL or JMS), which supports XA, then they would work in tandem,
where they both will either commit or rollback at the end of the exchange
routing. In some cases, if the JTA transaction manager fails to commit (during
the 2PC processing), but kafka transaction has been committed before and there
is no chance to roll back the changes since the kafka transaction does not
support JTA/XA spec. There is still a risk with the data consistency.
-When you use the `transacted=true` parameter, the `transactionalId` is made
from the endpoint id and route id, but if you want finer control you can just
use the `transactionalId` parameter, however remeber to have a
`transactionalId` for each kafka producer endpoint. Note that the old style of
using `additionalProperties[transactional.id]` is still valid.
+When you use the `transacted=true` parameter, the `transactionalId` is made
from the endpoint ID and route ID.
+If you want finer control, set the `transactionalId` parameter; use a
different ID for each producer endpoint.
+The old style of using `additionalProperties[transactional.id]` is still
valid. For exactly-once read-process-write,
+the ID must be unique across application replicas and stable across restarts
of the same producer instance; the
+generated endpoint/route ID alone does not meet that requirement.
If both `transacted=true` and `transactionalId` are present, the latter takes
precedence.
+=== Kafka exactly-once (read-process-write)
+
+The Kafka component can atomically commit records produced by a route and the
source consumer offsets. This is an
+opt-in feature for Kafka-to-Kafka routes. Configure `exactlyOnce=true` on both
Kafka endpoints:
+
+* Source consumer: `exactlyOnce=true`, `allowManualCommit=true`,
`autoCommitEnable=false` and
+ `breakOnFirstError=true`. Automatic commits are disabled. A transactional
destination producer commits the source
+ offsets with its output records; when a successful route branch does not use
a Kafka producer (for example, a filter
+ drops the record), Camel commits the source offset directly on the polling
consumer thread.
+* Destination producer: `exactlyOnce=true` and an explicit `transactionalId`
+ (or `additionalProperties[transactional.id]`).
+ `transacted=true` alone is not sufficient in this mode. Treat the configured
`transactionalId` as a base ID: Camel
+ appends the owning route ID when it creates the producer
(`<base>-<routeId>`). This gives separate producer instances
+ created for the same sink URI in different routes distinct Kafka
transactional IDs. Configure an explicit, stable
+ `routeId` for each route, and choose a base `transactionalId` that is unique
to this Camel runtime instance and
+ stable across restarts. The base ID must still differ across application
replicas; the route suffix does not make two
+ different Camel instances globally unique. If the same route creates
multiple independent producer instances with the
+ same base ID, configure distinct base IDs for those producers as well.
+* Consumers reading the destination topic: `isolation.level=read_committed`.
+
+[source,java]
+----
+from("kafka:in?groupId=my-group&exactlyOnce=true&allowManualCommit=true&autoCommitEnable=false&breakOnFirstError=true")
+
.to("kafka:out?exactlyOnce=true&transactionalId={{orders.kafka.transactional-id}}");
+----
+
+The manual-commit token is used by Camel to capture source offset metadata; do
not call its `commit()` method in
+this mode. The consumer group metadata is captured on the Kafka polling thread
and retained in an exchange property,
+so `removeHeaders("*")` or replacing the message does not discard the source
offsets. The producer sends the offsets
+to the transaction with `sendOffsetsToTransaction` immediately before
`commitTransaction`. The metadata must belong
+to one consumer group generation; a rebalance before the transaction commits
causes Kafka to reject the stale
+generation and the transaction is recovered as a failure. Kafka brokers must
be version 2.5 or later to support the
+full `ConsumerGroupMetadata` fencing information.
+
+The source endpoint must use an explicit, non-blank `groupId`, must not
configure `offsetRepository`, and must use
+`consumersCount=1`. The built-in synchronous and asynchronous manual-commit
factories are supported. A custom
+`KafkaManualCommitFactory` is not supported because Camel needs the record
offset, leader epoch and poll-thread-captured
+group metadata. `batching=true`, `consumersCount` greater than one, `seekTo`,
and a `ResumeStrategy` are rejected on the
+source: batching can span consumer group generations, while seek/resume
policies can reposition the consumer past
+uncommitted records. Configure the source and destination endpoints against
the *same Kafka cluster*; cross-cluster
+offset commits are not supported. Streaming mode currently runs one source
record per transaction.
+For throughput, tune the Kafka producer's `batch.size` and `linger.ms`
options; do not enable consumer `batching`
+with exactly-once mode. To publish to several output topics in one
transaction, use one exactly-once producer endpoint
+and set the `CamelKafkaOverrideTopic` header for each output record rather
than routing to multiple Kafka producer
+endpoints.
+
+`breakOnFirstError=true` is required so a failed record is replayed from the
last committed offset. If the input topic
+is itself written transactionally, configure `isolation.level=read_committed`
on the source consumer as well.
+
+Exactly-once mode supports one Kafka source and one exactly-once producer
transaction for the normal source route.
+A Kafka producer used by the normal route must set `exactlyOnce=true`; a
non-transactional or ordinary transactional
+destination is rejected as a source-claim failure when the exchange is
processed, not during route startup. This is
+not a Kafka send failure: even if an error handler consumes the rejection, the
source transaction is aborted and the
+source record is replayed rather than skipped. A Kafka producer used as the
destination of the selected
+failure handler is supported as a separate case. This includes a
`deadLetterChannel` destination and
+an `onException(...).handled(true)` destination. Sends after a `doCatch`
block, as well as sends in the
+original route once processing resumes following an
`onException(...).continued(true)` clause, are
+normal route sends, not failure-handler deliveries. If a
Review Comment:
💡 **Doc gap: sends *inside* a `doCatch` block behave differently from sends
*after* it**
The current text covers sends *after* the `doCatch` block ends ("are normal
route sends"). It does not address sends *inside* the `doCatch` body. With the
current implementation, sends inside a `doCatch` body have
`FAILURE_HANDLER_DELIVERY_PROPERTY=true` (because `CatchProcessor` fires
`ExchangeFailureHandlingEvent` around its body). This means Kafka sends inside
`doCatch` are classified as failure-handler deliveries — not as normal sends.
If this is the intended behavior, the documentation should be updated to
reflect it. If not, see the inline comment on the notifier code.
```suggestion
an `onException(...).handled(true)` destination. Sends after a `doCatch`
block, as well as sends in the
original route once processing resumes following an
`onException(...).continued(true)` clause, are
normal route sends, not failure-handler deliveries. Sends *inside* a
`doCatch` body are classified as
failure-handler deliveries and follow the failure-handler path. If a
```
--
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]
