[
https://issues.apache.org/jira/browse/CAMEL-23937?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18094889#comment-18094889
]
Abhishek Mittal commented on CAMEL-23937:
-----------------------------------------
[~davsclaus] Thanks for fixing this.
Do we have any plan to backport these change to 4.14.X or 4.18.X versions?
> camel-azure-servicebus consumer defeats SDK auto lock-renewal for async
> routes, causing silent message-lock expiry mid-processing
> ---------------------------------------------------------------------------------------------------------------------------------
>
> Key: CAMEL-23937
> URL: https://issues.apache.org/jira/browse/CAMEL-23937
> Project: Camel
> Issue Type: Bug
> Components: camel-azure-servicebus
> Affects Versions: 4.14.5
> Reporter: Abhishek Mittal
> Assignee: Claus Ibsen
> Priority: Major
> Fix For: 4.21.0
>
>
> The Service Bus consumer processes exchanges asynchronously but the
> underlying SDK's MessagePump disposes the lock-renewal subscription
> synchronously as soon as Camel's
> callback returns — not when the exchange is actually settled. Because
> Camel's consumer returns immediately for async routes, lock auto-renewal is
> cancelled almost immediately
> after message receipt, regardless of the configured
> maxAutoLockRenewDuration.
> Camel side —
> org.apache.camel.component.azure.servicebus.ServiceBusConsumer#processMessage
> (ServiceBusConsumer.java:77-86):
> private void processMessage(ServiceBusReceivedMessageContext messageContext)
> { final ServiceBusReceivedMessage message =
> messageContext.getMessage(); final Exchange exchange =
> createServiceBusExchange(message); final ConsumerOnCompletion
> onCompletion = new ConsumerOnCompletion(messageContext); // add
> exchange callback
> exchange.getExchangeExtension().addOnCompletion(onCompletion); // use
> default consumer callback AsyncCallback cb =
> defaultConsumerCallback(exchange, true);
> getAsyncProcessor().process(exchange, cb); }
> getAsyncProcessor().process(exchange, cb) schedules the route and returns
> immediately; actual settlement (complete()/abandon()/deadLetter()) happens
> later, from the
> ConsumerOnCompletion Synchronization (ServiceBusConsumer.java:165-206) once
> the exchange finishes — which may be long after this method returns.
>
> org.apache.camel.component.azure.servicebus.client.ServiceBusClientFactory#createBaseServiceBusProcessorClient
> (ServiceBusClientFactory.java:65-80, mirrored in
> createBaseServiceBusSessionProcessorClient, lines 82-98) explicitly
> disables the SDK's own auto-complete because Camel manages settlement itself:
> // We handle auto-complete in the consumer, since we have no way to
> propagate errors back to the reactive
> // pipeline messages are published on so the message would be completed
> even if an error occurs during Exchange
> // processing.
> processorClientBuilder.disableAutoComplete();
> maxAutoLockRenewDuration is still passed through to the SDK builder
> (ServiceBusClientFactory.java:114 / 134), giving the impression that lock
> renewal is configured for the full expected processing window.
> SDK side — com.azure.messaging.servicebus.MessagePump#handleMessage
> (MessagePump.java:140-159):
> private void handleMessage(ServiceBusReceivedMessage message) {
> instrumentation.instrumentProcess(message, ReceiverKind.PROCESSOR, msg
> -> {
> final Disposable lockRenewDisposable;
> if (enableAutoLockRenew)
> { lockRenewDisposable = client.beginLockRenewal(message);
> }
> else
> { lockRenewDisposable = Disposables.disposed(); }
> final Throwable error = notifyMessage(message);
> if (enableAutoDisposition) {
> if (error == null)
> { complete(message); }
> else
> { abandon(message); }
> }
> lockRenewDisposable.dispose();
> return error;
> });
> }
> notifyMessage(message) (MessagePump.java:161-170) synchronously invokes the
> Camel processMessage callback above and returns as soon as that call returns
> — which, for an async Camel route, is almost immediately, well before the
> exchange is settled. The very next line unconditionally disposes
> lockRenewDisposable, cancelling auto-renewal at that point,
> not when processing actually completes. Since enableAutoDisposition is
> false here (Camel called disableAutoComplete()), the complete()/abandon()
> branch is skipped too —
> settlement is left entirely to Camel's later, out-of-band
> ConsumerOnCompletion callback, by which time the lock has already lost its
> renewal and may have expired.
> Net effect: renewal is torn down roughly at t+(sync callback duration),
> typically a few milliseconds, irrespective of maxAutoLockRenewDuration. Any
> route whose actual
> processing time exceeds the entity's LockDuration will have its lock expire
> while still "in flight" from Camel's perspective.
> *Steps to reproduce*
> 1. Configure a camel-azure-servicebus consumer endpoint on a topic with
> PEEK_LOCK receive mode and a short LockDuration (e.g. 30s) on the Service Bus
> entity.
> 2. Set maxAutoLockRenewDuration to a large value (e.g. 5 minutes) expecting
> it to keep the lock alive.
> 3. Route processing (async processor, aggregator, throttler, idempotent
> consumer, or simply business logic) takes longer than 30s.
> 4. Observe: the lock is not renewed past the initial LockDuration window;
> the broker redelivers the message to another receiver while the first
> exchange is still processing,
> and/or the eventual complete()/abandon() call from ConsumerOnCompletion
> fails or becomes a no-op against the now-invalid lock token, with no error
> surfaced to the application.
> A standalone repro project (minimal Camel route + Azure Service Bus
> emulator/live namespace) is available on request.
> *Expected behavior*
> With maxAutoLockRenewDuration configured, the SDK should keep renewing the
> message lock for the full duration the exchange is actually being processed
> by Camel (i.e., until
> ConsumerOnCompletion.onComplete/onFailure fires), up to the configured
> maximum — not just for the duration of the synchronous callback invocation.
> *Actual behavior*
> Lock renewal is disposed essentially immediately after the synchronous
> processMessage callback returns, because MessagePump.handleMessage ties
> lockRenewDisposable.dispose() to
> the callback's return rather than to actual exchange completion. For async
> Camel routes this happens long before processing finishes, so the lock
> silently expires
> mid-processing.
> *Impact / severity*
> Data loss / duplicate delivery, silent. Once the lock expires, the broker
> may redeliver the message to a different receiver while the original exchange
> is still being
> processed. The original exchange's later complete() (via
> ConsumerOnCompletion.onComplete) becomes a no-op against the expired lock, so:
> - the message may be lost if the redelivered copy is also not properly
> settled, or
> - the message may be processed twice (original + redelivered copy), or
> - in dead-lettering/abandon paths, settlement calls fail silently.
> No exception is surfaced to the Camel route or the application in the
> common case — this can pass light/functional testing (where processing is
> fast) and only surfaces in
> production under load or with any processing step that pushes total
> handling time past LockDuration.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
