davsclaus commented on code in PR #28730:
URL: https://github.com/apache/camel/pull/28730#discussion_r4452309976
##########
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:
Addressed in a7de0ba5b2: the docs and the notifier comment now say that
Kafka sends inside a `doCatch` body take the failure-handler delivery path, and
`KafkaExactlyOnceIT` pins that. Resolving.
_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]
