This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 45034f355d1c CAMEL-23937: Add Camel-managed lock renewal for Azure
Service Bus async routes
45034f355d1c is described below
commit 45034f355d1c29e4d208fa31b9e285c6948551b5
Author: Claus Ibsen <[email protected]>
AuthorDate: Wed Jul 8 22:29:17 2026 +0200
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_22.adoc | 18 +++
4 files changed, 310 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_22.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
index 02f1788a73b9..d6a7b9e751b7 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
@@ -69,6 +69,24 @@ to the true average throughput rather than oscillating
between zero and spike va
If you have tooling that consumes the JMX throughput value and expects the old
instantaneous
behavior, be aware that the reported value will now ramp up and decay
gradually.
+=== 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.
+
=== camel-debezium
The `camel-debezium` component has upgraded from Debezium 3.5.2 to 3.6.0.
