ramu11 commented on code in PR #28730:
URL: https://github.com/apache/camel/pull/28730#discussion_r4442023651
##########
components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/streaming/KafkaRecordStreamingProcessor.java:
##########
@@ -104,6 +145,363 @@ 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 actually performed by this source
exchange. The event notifier marks the shared
+ * metadata object, so the original source observes a hand-off even when
Camel emits the event for a copied
+ * exchange.
+ */
+ boolean hasAsynchronousHandoff(Exchange exchange) {
+ String toEndpoint =
exchange.getProperty(ExchangePropertyKey.TO_ENDPOINT, String.class);
+ KafkaExactlyOnceMetadata metadata = exchange.getProperty(
+ KafkaExactlyOnceMetadata.EXCHANGE_PROPERTY,
KafkaExactlyOnceMetadata.class);
+ return isAsynchronousHandoffEndpoint(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;
+ }
+
+ 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;
+ }
+ if
(isAsynchronousHandoffEndpoint(sendingEvent.getEndpoint().getEndpointUri())) {
+ // The metadata object is shared by shallow
exchange copies; do not put this marker only on
+ // the copied exchange that raised
ExchangeSendingEvent.
+ metadata.markAsynchronousKafkaHandoff();
+ }
+ return;
+ }
+
+ if (event instanceof
CamelEvent.ExchangeFailureHandlingEvent handlingEvent) {
+ Exchange exchange = handlingEvent.getExchange();
+ if (metadata(exchange) == null ||
!handlingEvent.isDeadLetterChannel()) {
+ 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 ||
!handledEvent.isDeadLetterChannel()) {
+ 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; keep them enabled to distinguish
+ // a real Dead Letter Channel from doTry/doCatch and
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));
+ }
+ }
+
+ /** 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 KafkaExactlyOnceMetadata metadata(Exchange exchange) {
+ return
exchange.getProperty(KafkaExactlyOnceMetadata.EXCHANGE_PROPERTY,
KafkaExactlyOnceMetadata.class);
+ }
+
+ /**
+ * Validate WireTap targets before a source consumer starts polling. A
WireTap runs on a copied exchange and cannot
+ * participate in the source transaction. Static target graphs are
followed through route endpoints; dynamic WireTap
+ * URIs and dynamic send EIPs in a target graph are rejected
conservatively because their destination is unknowable
+ * at startup.
+ */
+ static void validateNoKafkaWireTaps(CamelContext context, String routeId) {
+ if (context == null || routeId == null) {
+ throw new IllegalArgumentException(
+ "exactlyOnce requires a route id and CamelContext to
validate WireTap destinations");
+ }
+ Route sourceRoute = context.getRoute(routeId);
+ if (sourceRoute == null) {
+ throw new IllegalArgumentException(
+ "exactlyOnce cannot validate WireTap destinations because
route " + routeId + " is not available");
+ }
+ if (containsKafkaWireTap(sourceRoute.navigate(), context, new
HashSet<>())) {
+ throw new IllegalArgumentException(
+ "Kafka exactlyOnce does not support a WireTap whose target
can reach a Kafka producer; "
+ + "send to Kafka synchronously
from the original source exchange (route "
+ + routeId + ")");
+ }
+ }
+
+ private static boolean containsKafkaWireTap(
+ Navigate<Processor> navigation, CamelContext context, Set<String>
visitedRoutes) {
+ if (navigation == null || !navigation.hasNext()) {
+ return false;
+ }
+ List<Processor> children = navigation.next();
+ if (children == null) {
+ return false;
+ }
+ for (Processor child : children) {
+ Processor processor = unwrapChannel(child);
+ if (processor == null) {
+ continue;
+ }
+ if (processor instanceof WireTapProcessor wireTap
+ && (isDynamicKafkaWireTap(wireTap) ||
routesToKafka(context, wireTap.getUri(), visitedRoutes))) {
+ return true;
+ }
+ if (processor instanceof Navigate<?> nested) {
+ @SuppressWarnings("unchecked")
+ Navigate<Processor> childNavigation = (Navigate<Processor>)
nested;
+ if (containsKafkaWireTap(childNavigation, context,
visitedRoutes)) {
+ return true;
+ }
+ }
+ }
+ return false;
+ }
+
+ private static boolean routesToKafka(CamelContext context, String uri,
Set<String> visitedRoutes) {
+ String target = endpointKey(uri);
+ if (target.startsWith("kafka:")) {
+ return true;
+ }
+ if (target.isEmpty() || !visitedRoutes.add(target)) {
+ return false;
+ }
+ for (Route route : context.getRoutes()) {
+ if (route.getEndpoint() != null
+ &&
endpointKey(route.getEndpoint().getEndpointUri()).equals(target)
+ && routeMayReachKafka(route.navigate(), context,
visitedRoutes)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private static boolean routeMayReachKafka(
+ Navigate<Processor> navigation, CamelContext context, Set<String>
visitedRoutes) {
+ if (navigation == null || !navigation.hasNext()) {
+ return false;
+ }
+ List<Processor> children = navigation.next();
+ if (children == null) {
+ return false;
+ }
+ for (Processor child : children) {
+ Processor processor = unwrapChannel(child);
+ if (processor == null) {
+ continue;
+ }
+ if (processor instanceof WireTapProcessor wireTap) {
+ if (isDynamicKafkaWireTap(wireTap) || routesToKafka(context,
wireTap.getUri(), visitedRoutes)) {
+ return true;
+ }
+ } else if (processor instanceof EndpointSending && !(processor
instanceof SendProcessor)) {
+ // SendDynamicProcessor, RecipientList and RoutingSlip can
resolve destinations at runtime.
+ return true;
+ } else if (processor instanceof EndpointAware endpointAware &&
endpointAware.getEndpoint() != null) {
+ String endpointUri =
endpointAware.getEndpoint().getEndpointUri();
+ String endpoint = endpointKey(endpointUri);
+ if (endpoint.startsWith("kafka:")) {
+ return true;
+ }
+ if (isRoutedEndpoint(endpoint) && routesToKafka(context,
endpointUri, visitedRoutes)) {
+ return true;
+ }
+ } else if (processor instanceof EndpointSending) {
+ // A dynamic send can resolve to Kafka at runtime; fail safe
for Kafka-directed WireTap graphs.
+ return true;
+ }
+
+ if (processor instanceof Navigate<?> nested) {
+ @SuppressWarnings("unchecked")
+ Navigate<Processor> childNavigation = (Navigate<Processor>)
nested;
+ if (routeMayReachKafka(childNavigation, context,
visitedRoutes)) {
+ return true;
+ }
+ }
+ }
+ return false;
+ }
+
+ /**
+ * Determine whether the WireTap actually uses a dynamic sender.
+ *
+ * The dynamicUri option defaults to true, but a static URI without a
Simple expression is still optimized to a
+ * normal SendProcessor by WireTapReifier. ignoreInvalidEndpoint also
requires a dynamic sender.
+ */
+ private static boolean isDynamicKafkaWireTap(WireTapProcessor wireTap) {
+ return wireTap.isIgnoreInvalidEndpoint()
+ || wireTap.isDynamicUri() &&
LanguageSupport.hasSimpleFunction(wireTap.getUri())
+ && LanguageSupport.hasSimpleFunction(wireTap.getUri());
Review Comment:
addressed
--
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]
