This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch camel-4.18.x
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/camel-4.18.x by this push:
new d3f5cdf91f9f [backport camel-4.18.x] CAMEL-23937: Add Camel-managed
lock renewal for Azure Service Bus async routes (#24559)
d3f5cdf91f9f is described below
commit d3f5cdf91f9f56c39688370b865871f3f07b9722
Author: Claus Ibsen <[email protected]>
AuthorDate: Thu Jul 9 15:17:45 2026 +0200
[backport camel-4.18.x] CAMEL-23937: Add Camel-managed lock renewal for
Azure Service Bus async routes (#24559)
CAMEL-23937: Add Camel-managed lock renewal for Azure Service Bus async
routes
The Azure SDK's built-in lock renewal (maxAutoLockRenewDuration) is tied
to the processMessage callback duration, which returns immediately for
async Camel routes. This adds a LockRenewer that uses a
ServiceBusReceiverAsyncClient to periodically renew message locks for
in-flight exchanges in PEEK_LOCK mode. Integrates with Camel's
PeriodTaskScheduler for operational visibility. Session-enabled consumers
are excluded (follow-up in CAMEL-23938).
Closes #24511
Co-authored-by: Claude Opus 4.6 <[email protected]>
---
.../azure/servicebus/ServiceBusConsumer.java | 131 ++++++++++++++++++++
.../servicebus/client/ServiceBusClientFactory.java | 28 +++++
.../azure/servicebus/ServiceBusConsumerTest.java | 133 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_18.adoc | 20 ++++
4 files changed, 312 insertions(+)
diff --git
a/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusConsumer.java
b/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusConsumer.java
index c4b153288f5a..ffe1c1af1c83 100644
---
a/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusConsumer.java
+++
b/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusConsumer.java
@@ -16,8 +16,13 @@
*/
package org.apache.camel.component.azure.servicebus;
+import java.time.Duration;
+import java.time.Instant;
+import java.time.OffsetDateTime;
import java.util.Arrays;
import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
@@ -25,6 +30,7 @@ import com.azure.messaging.servicebus.ServiceBusErrorContext;
import com.azure.messaging.servicebus.ServiceBusProcessorClient;
import com.azure.messaging.servicebus.ServiceBusReceivedMessage;
import com.azure.messaging.servicebus.ServiceBusReceivedMessageContext;
+import com.azure.messaging.servicebus.ServiceBusReceiverAsyncClient;
import com.azure.messaging.servicebus.models.DeadLetterOptions;
import com.azure.messaging.servicebus.models.ServiceBusReceiveMode;
import com.azure.messaging.servicebus.models.SubQueue;
@@ -34,8 +40,10 @@ import org.apache.camel.Message;
import org.apache.camel.Processor;
import org.apache.camel.ShutdownRunningTask;
import org.apache.camel.spi.HeaderFilterStrategy;
+import org.apache.camel.spi.PeriodTaskScheduler;
import org.apache.camel.spi.ShutdownAware;
import org.apache.camel.support.DefaultConsumer;
+import org.apache.camel.support.PluginHelper;
import org.apache.camel.support.SynchronizationAdapter;
import org.apache.camel.util.ObjectHelper;
import org.slf4j.Logger;
@@ -44,7 +52,11 @@ import org.slf4j.LoggerFactory;
public class ServiceBusConsumer extends DefaultConsumer implements
ShutdownAware {
private static final Logger LOG =
LoggerFactory.getLogger(ServiceBusConsumer.class);
+ private static final int LOCK_RENEW_INTERVAL_SECONDS = 10;
+
private ServiceBusProcessorClient client;
+ private ServiceBusReceiverAsyncClient renewalClient;
+ private LockRenewer lockRenewer;
private final AtomicInteger pendingExchanges = new AtomicInteger();
public ServiceBusConsumer(final ServiceBusEndpoint endpoint, final
Processor processor) {
@@ -76,6 +88,23 @@ public class ServiceBusConsumer extends DefaultConsumer
implements ShutdownAware
}
client.start();
+
+ // set up Camel-managed lock renewal for PEEK_LOCK mode
+ // the Azure SDK's built-in lock renewal is tied to the processMessage
callback duration,
+ // which returns immediately for async routes, making
maxAutoLockRenewDuration ineffective
+ if (getConfiguration().getProcessorClient() == null
+ && getConfiguration().getServiceBusReceiveMode() ==
ServiceBusReceiveMode.PEEK_LOCK
+ && getConfiguration().getMaxAutoLockRenewDuration() > 0
+ && !getConfiguration().isSessionEnabled()) {
+ renewalClient = getEndpoint().getServiceBusClientFactory()
+ .createServiceBusReceiverAsyncClient(getConfiguration());
+
+ long maxRenewMs = getConfiguration().getMaxAutoLockRenewDuration();
+ lockRenewer = new LockRenewer(renewalClient,
Duration.ofMillis(maxRenewMs));
+
+ PeriodTaskScheduler scheduler =
PluginHelper.getPeriodTaskScheduler(getEndpoint().getCamelContext());
+ scheduler.schedulePeriodTask(lockRenewer,
LOCK_RENEW_INTERVAL_SECONDS * 1000L);
+ }
}
private void processMessage(ServiceBusReceivedMessageContext
messageContext) {
@@ -85,6 +114,10 @@ public class ServiceBusConsumer extends DefaultConsumer
implements ShutdownAware
final ConsumerOnCompletion onCompletion = new
ConsumerOnCompletion(messageContext);
// add exchange callback
exchange.getExchangeExtension().addOnCompletion(onCompletion);
+ // track for Camel-managed lock renewal
+ if (lockRenewer != null) {
+ lockRenewer.add(exchange, message);
+ }
// use default consumer callback
AsyncCallback cb = defaultConsumerCallback(exchange, true);
getAsyncProcessor().process(exchange, cb);
@@ -102,6 +135,14 @@ public class ServiceBusConsumer extends DefaultConsumer
implements ShutdownAware
@Override
protected void doStop() throws Exception {
+ if (lockRenewer != null) {
+ lockRenewer.cancel();
+ lockRenewer = null;
+ }
+ if (renewalClient != null) {
+ renewalClient.close();
+ renewalClient = null;
+ }
if (client != null) {
// stop accepting new messages but keep the connection open
// so that in-flight exchanges can still complete/abandon messages
@@ -114,6 +155,15 @@ public class ServiceBusConsumer extends DefaultConsumer
implements ShutdownAware
@Override
protected void doShutdown() throws Exception {
+ // defensive cleanup in case doStop() was not called
+ if (lockRenewer != null) {
+ lockRenewer.cancel();
+ lockRenewer = null;
+ }
+ if (renewalClient != null) {
+ renewalClient.close();
+ renewalClient = null;
+ }
if (client != null) {
// close the client after all in-flight exchanges have completed
client.close();
@@ -198,6 +248,87 @@ public class ServiceBusConsumer extends DefaultConsumer
implements ShutdownAware
return exchange;
}
+ private class LockRenewer implements Runnable {
+
+ private final ServiceBusReceiverAsyncClient client;
+ private final Duration maxRenewDuration;
+ private final AtomicBoolean run = new AtomicBoolean(true);
+ private final Map<String, LockRenewerEntry> entries = new
ConcurrentHashMap<>();
+
+ LockRenewer(ServiceBusReceiverAsyncClient client, Duration
maxRenewDuration) {
+ this.client = client;
+ this.maxRenewDuration = maxRenewDuration;
+ }
+
+ public void add(Exchange exchange, ServiceBusReceivedMessage message) {
+ exchange.getExchangeExtension().addOnCompletion(new
SynchronizationAdapter() {
+ @Override
+ public void onComplete(Exchange exchange) {
+ remove(exchange);
+ }
+
+ @Override
+ public void onFailure(Exchange exchange) {
+ remove(exchange);
+ }
+
+ private void remove(Exchange exchange) {
+ LOG.trace("Removing exchangeId {} from the LockRenewer,
processing done", exchange.getExchangeId());
+ entries.remove(exchange.getExchangeId());
+ }
+ });
+
+ entries.put(exchange.getExchangeId(),
+ new LockRenewerEntry(message, message.getLockedUntil(),
Instant.now()));
+ }
+
+ public void cancel() {
+ run.set(false);
+ }
+
+ @Override
+ public void run() {
+ if (!run.get()) {
+ return;
+ }
+
+ Instant now = Instant.now();
+ for (Map.Entry<String, LockRenewerEntry> mapEntry :
entries.entrySet()) {
+ String exchangeId = mapEntry.getKey();
+ LockRenewerEntry entry = mapEntry.getValue();
+
+ // skip if max renewal duration exceeded
+ Duration elapsed = Duration.between(entry.startTime, now);
+ if (elapsed.compareTo(maxRenewDuration) >= 0) {
+ LOG.debug("Max lock renewal duration exceeded for
exchangeId {}, stopping renewal", exchangeId);
+ entries.remove(exchangeId);
+ continue;
+ }
+
+ // renew if lock is approaching expiry
+ Instant lockExpiry = entry.lockedUntil.toInstant();
+ if
(now.plusSeconds(LOCK_RENEW_INTERVAL_SECONDS).isAfter(lockExpiry)) {
+ client.renewMessageLock(entry.message)
+ .subscribe(
+ newLockedUntil -> {
+ LOG.trace("Renewed lock for exchangeId
{}, new expiry: {}", exchangeId,
+ newLockedUntil);
+ entries.computeIfPresent(exchangeId,
+ (k, e) -> new
LockRenewerEntry(e.message, newLockedUntil, e.startTime));
+ },
+ error -> {
+ LOG.warn("Failed to renew lock for
exchangeId {}: {}", exchangeId,
+ error.getMessage());
+ entries.remove(exchangeId);
+ });
+ }
+ }
+ }
+
+ private record LockRenewerEntry(ServiceBusReceivedMessage message,
OffsetDateTime lockedUntil, Instant startTime) {
+ }
+ }
+
private class ConsumerOnCompletion extends SynchronizationAdapter {
private final ServiceBusReceivedMessageContext messageContext;
diff --git
a/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/client/ServiceBusClientFactory.java
b/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/client/ServiceBusClientFactory.java
index 84070214d6ce..91ea8ab77d41 100644
---
a/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/client/ServiceBusClientFactory.java
+++
b/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/client/ServiceBusClientFactory.java
@@ -97,11 +97,39 @@ public final class ServiceBusClientFactory {
return processorClientBuilder;
}
+ private static ServiceBusClientBuilder.ServiceBusReceiverClientBuilder
createBaseServiceBusReceiverClient(
+ final ServiceBusClientBuilder busClientBuilder, final
ServiceBusConfiguration configuration) {
+ final ServiceBusClientBuilder.ServiceBusReceiverClientBuilder
receiverClientBuilder = busClientBuilder.receiver();
+
+ receiverClientBuilder.disableAutoComplete();
+
+ switch (configuration.getServiceBusType()) {
+ case queue ->
receiverClientBuilder.queueName(configuration.getTopicOrQueueName());
+ case topic ->
receiverClientBuilder.topicName(configuration.getTopicOrQueueName());
+ }
+
+ return receiverClientBuilder;
+ }
+
public ServiceBusSenderClient createServiceBusSenderClient(final
ServiceBusConfiguration configuration) {
return
createBaseServiceBusSenderClient(createBaseServiceBusClient(configuration),
configuration)
.buildClient();
}
+ public ServiceBusReceiverAsyncClient
createServiceBusReceiverAsyncClient(final ServiceBusConfiguration
configuration) {
+ ServiceBusClientBuilder.ServiceBusReceiverClientBuilder clientBuilder
+ =
createBaseServiceBusReceiverClient(createBaseServiceBusClient(configuration),
configuration);
+
+ clientBuilder
+ .subscriptionName(configuration.getSubscriptionName())
+ .receiveMode(configuration.getServiceBusReceiveMode())
+ .maxAutoLockRenewDuration(Duration.ZERO)
+ .prefetchCount(configuration.getPrefetchCount())
+ .subQueue(configuration.getSubQueue());
+
+ return clientBuilder.buildAsyncClient();
+ }
+
public ServiceBusProcessorClient createServiceBusProcessorClient(
ServiceBusConfiguration configuration,
Consumer<ServiceBusReceivedMessageContext> processMessage,
Consumer<ServiceBusErrorContext> processError) {
diff --git
a/components/camel-azure/camel-azure-servicebus/src/test/java/org/apache/camel/component/azure/servicebus/ServiceBusConsumerTest.java
b/components/camel-azure/camel-azure-servicebus/src/test/java/org/apache/camel/component/azure/servicebus/ServiceBusConsumerTest.java
index 0965859c715f..7153a3325c1f 100644
---
a/components/camel-azure/camel-azure-servicebus/src/test/java/org/apache/camel/component/azure/servicebus/ServiceBusConsumerTest.java
+++
b/components/camel-azure/camel-azure-servicebus/src/test/java/org/apache/camel/component/azure/servicebus/ServiceBusConsumerTest.java
@@ -34,7 +34,9 @@ import
org.apache.camel.component.azure.servicebus.client.ServiceBusClientFactor
import org.apache.camel.spi.ExceptionHandler;
import org.apache.camel.spi.ExchangeFactory;
import org.apache.camel.spi.HeaderFilterStrategy;
+import org.apache.camel.spi.PeriodTaskScheduler;
import org.apache.camel.spi.Synchronization;
+import org.apache.camel.spi.UuidGenerator;
import org.apache.camel.support.DefaultExchange;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -44,6 +46,7 @@ import org.mockito.ArgumentCaptor;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.Mockito.*;
public class ServiceBusConsumerTest {
@@ -87,6 +90,9 @@ public class ServiceBusConsumerTest {
private final ExchangeFactory ef = mock();
private final HeaderFilterStrategy headerFilterStrategy = mock();
private final ExceptionHandler exceptionHandler = mock();
+ private final PeriodTaskScheduler periodTaskScheduler = mock();
+ private final UuidGenerator uuidGenerator = mock();
+ private final ServiceBusReceiverAsyncClient renewalClient = mock();
private final ArgumentCaptor<Consumer<ServiceBusReceivedMessageContext>>
processMessageCaptor = ArgumentCaptor.captor();
private final ArgumentCaptor<Consumer<ServiceBusErrorContext>>
processErrorCaptor = ArgumentCaptor.captor();
private final ArgumentCaptor<Exchange> exchangeCaptor =
ArgumentCaptor.captor();
@@ -110,6 +116,10 @@ public class ServiceBusConsumerTest {
.thenReturn(client);
when(processor.process(exchangeCaptor.capture(),
asyncCallbackCaptor.capture())).thenReturn(true);
when(configuration.getHeaderFilterStrategy()).thenReturn(headerFilterStrategy);
+
when(ecc.getContextPlugin(PeriodTaskScheduler.class)).thenReturn(periodTaskScheduler);
+ when(context.getUuidGenerator()).thenReturn(uuidGenerator);
+
when(uuidGenerator.generateExchangeUuid()).thenReturn("test-exchange-id");
+
when(clientFactory.createServiceBusReceiverAsyncClient(any())).thenReturn(renewalClient);
}
@Test
@@ -507,6 +517,129 @@ public class ServiceBusConsumerTest {
}
}
+ @Test
+ void lockRenewerCreatedForPeekLockMode() throws Exception {
+
when(configuration.getServiceBusReceiveMode()).thenReturn(ServiceBusReceiveMode.PEEK_LOCK);
+ when(configuration.getMaxAutoLockRenewDuration()).thenReturn(300000L);
+ try (ServiceBusConsumer consumer = new ServiceBusConsumer(endpoint,
processor)) {
+ consumer.doStart();
+ verify(clientFactory).createServiceBusReceiverAsyncClient(any());
+ verify(periodTaskScheduler).schedulePeriodTask(any(), anyLong());
+ }
+ }
+
+ @Test
+ void lockRenewerNotCreatedForReceiveAndDeleteMode() throws Exception {
+
when(configuration.getServiceBusReceiveMode()).thenReturn(ServiceBusReceiveMode.RECEIVE_AND_DELETE);
+ when(configuration.getMaxAutoLockRenewDuration()).thenReturn(300000L);
+ try (ServiceBusConsumer consumer = new ServiceBusConsumer(endpoint,
processor)) {
+ consumer.doStart();
+ verify(clientFactory,
never()).createServiceBusReceiverAsyncClient(any());
+ }
+ }
+
+ @Test
+ void lockRenewerNotCreatedWhenMaxAutoLockRenewDurationIsZero() throws
Exception {
+
when(configuration.getServiceBusReceiveMode()).thenReturn(ServiceBusReceiveMode.PEEK_LOCK);
+ when(configuration.getMaxAutoLockRenewDuration()).thenReturn(0L);
+ try (ServiceBusConsumer consumer = new ServiceBusConsumer(endpoint,
processor)) {
+ consumer.doStart();
+ verify(clientFactory,
never()).createServiceBusReceiverAsyncClient(any());
+ }
+ }
+
+ @Test
+ void lockRenewerNotCreatedForSessionEnabledConsumer() throws Exception {
+ when(configuration.isSessionEnabled()).thenReturn(true);
+
when(configuration.getServiceBusReceiveMode()).thenReturn(ServiceBusReceiveMode.PEEK_LOCK);
+ when(configuration.getMaxAutoLockRenewDuration()).thenReturn(300000L);
+ try (ServiceBusConsumer consumer = new ServiceBusConsumer(endpoint,
processor)) {
+ consumer.doStart();
+ verify(clientFactory,
never()).createServiceBusReceiverAsyncClient(any());
+ }
+ }
+
+ @Test
+ void lockRenewerNotCreatedWhenCustomProcessorClientProvided() throws
Exception {
+ ServiceBusProcessorClient customClient = mock();
+ when(configuration.getProcessorClient()).thenReturn(customClient);
+
when(configuration.getServiceBusReceiveMode()).thenReturn(ServiceBusReceiveMode.PEEK_LOCK);
+ when(configuration.getMaxAutoLockRenewDuration()).thenReturn(300000L);
+ try (ServiceBusConsumer consumer = new ServiceBusConsumer(endpoint,
processor)) {
+ consumer.doStart();
+ verify(clientFactory,
never()).createServiceBusReceiverAsyncClient(any());
+ }
+ }
+
+ @Test
+ void lockRenewerTracksAndRemovesExchanges() throws Exception {
+
when(configuration.getServiceBusReceiveMode()).thenReturn(ServiceBusReceiveMode.PEEK_LOCK);
+ when(configuration.getMaxAutoLockRenewDuration()).thenReturn(300000L);
+ when(messageContext.getMessage()).thenReturn(message);
+ // lock expiring soon so the renewer would attempt renewal
+
when(message.getLockedUntil()).thenReturn(OffsetDateTime.now().plusSeconds(5));
+ try (ServiceBusConsumer consumer = new ServiceBusConsumer(endpoint,
processor)) {
+ consumer.doStart();
+
+ // capture the lock renewer task scheduled with PeriodTaskScheduler
+ ArgumentCaptor<Runnable> taskCaptor = ArgumentCaptor.captor();
+
verify(periodTaskScheduler).schedulePeriodTask(taskCaptor.capture(), anyLong());
+ Runnable lockRenewerTask = taskCaptor.getValue();
+
+ processMessageCaptor.getValue().accept(messageContext);
+
+ // complete the exchange — should remove from lock renewer tracking
+ Exchange exchange = exchangeCaptor.getValue();
+ for (Synchronization sync :
exchange.getExchangeExtension().handoverCompletions()) {
+ sync.onComplete(exchange);
+ }
+
+ // run the lock renewer — it should NOT attempt renewal for the
completed exchange
+ lockRenewerTask.run();
+ verify(renewalClient,
never()).renewMessageLock(any(ServiceBusReceivedMessage.class));
+ }
+ }
+
+ @Test
+ void doStopCleansUpLockRenewer() throws Exception {
+
when(configuration.getServiceBusReceiveMode()).thenReturn(ServiceBusReceiveMode.PEEK_LOCK);
+ when(configuration.getMaxAutoLockRenewDuration()).thenReturn(300000L);
+ ServiceBusConsumer consumer = new ServiceBusConsumer(endpoint,
processor);
+ consumer.doStart();
+
+ verify(clientFactory).createServiceBusReceiverAsyncClient(any());
+
+ consumer.doStop();
+
+ verify(renewalClient).close();
+ verify(client).stop();
+ verify(client, never()).close();
+ }
+
+ @Test
+ void stopStartCycleRecreatesLockRenewer() throws Exception {
+
when(configuration.getServiceBusReceiveMode()).thenReturn(ServiceBusReceiveMode.PEEK_LOCK);
+ when(configuration.getMaxAutoLockRenewDuration()).thenReturn(300000L);
+ ServiceBusConsumer consumer = new ServiceBusConsumer(endpoint,
processor);
+ consumer.doStart();
+
+ verify(clientFactory,
times(1)).createServiceBusReceiverAsyncClient(any());
+
+ consumer.doStop();
+ verify(renewalClient).close();
+
+ // second start should create a fresh renewal client
+ ServiceBusReceiverAsyncClient renewalClient2 = mock();
+
when(clientFactory.createServiceBusReceiverAsyncClient(any())).thenReturn(renewalClient2);
+ consumer.doStart();
+
+ verify(clientFactory,
times(2)).createServiceBusReceiverAsyncClient(any());
+ verify(periodTaskScheduler, times(2)).schedulePeriodTask(any(),
anyLong());
+
+ consumer.doShutdown();
+ verify(renewalClient2).close();
+ }
+
@Test
void customProcessorClient() throws Exception {
ServiceBusProcessorClient customClient = mock();
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_18.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_18.adoc
index 967557d80c3b..0358aaeba807 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_18.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_18.adoc
@@ -11,6 +11,26 @@ Note that manual migration is still required.
See the xref:camel-upgrade-recipes-tool.adoc[documentation] page for details.
====
+== Upgrading from 4.18.3 to 4.18.4
+
+=== camel-azure-servicebus - Camel-managed message lock renewal
+
+When consuming from Azure Service Bus in `PEEK_LOCK` mode with
`maxAutoLockRenewDuration > 0`, Camel now
+actively renews message locks using a dedicated
`ServiceBusReceiverAsyncClient` and Camel's internal
+`PeriodTaskScheduler`. This addresses a limitation where the Azure SDK's
built-in lock renewal is tied
+to the `processMessage` callback duration, which returns immediately for
asynchronous Camel routes —
+causing long-running exchanges to silently lose their message locks.
+
+The new behavior activates automatically when all of the following are true:
+
+* No custom `processorClient` is provided
+* Receive mode is `PEEK_LOCK`
+* `maxAutoLockRenewDuration > 0`
+* Session mode is disabled
+
+No configuration changes are required. The existing `maxAutoLockRenewDuration`
option controls how long
+Camel will continue renewing a message's lock.
+
== Upgrading from 4.18.2 to 4.18.3
=== camel-spring-ws - potential breaking change
