ramu11 commented on code in PR #28730:
URL: https://github.com/apache/camel/pull/28730#discussion_r4451391384
##########
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:
addressed
##########
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:
adressed
--
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]
