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

["Abhishek Mittal (Jira)" <ji...@apache.org>]

Abhishek Mittal created CAMEL-23937:
---------------------------------------

             Summary: 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


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)