davsclaus commented on code in PR #28730:
URL: https://github.com/apache/camel/pull/28730#discussion_r4451644297
##########
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:
Fixed in a7de0ba5b2: route events now clear the reachability cache, and
rechecksNonKafkaHandoffWhenRouteIsAddedAtRuntime covers a route added at
runtime. Resolving.
_Claude Code on behalf of davsclaus_
##########
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:
Fixed: kafka-component.adoc now says that sends inside a doCatch body are
failure-handler deliveries, and the catalog copy matches. Resolving. (The
classification itself is discussed in the other thread and in my re-review.)
_Claude Code on behalf of davsclaus_
--
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]
