[jira] [Assigned] (CAMEL-23937) camel-azure-servicebus consumer defeats SDK auto lock-renewal for async routes, causing silent message-lock expiry mid-processing

["Claus Ibsen (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/CAMEL-23937?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Claus Ibsen reassigned CAMEL-23937:
-----------------------------------

    Assignee: Claus Ibsen

> 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)