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)
