This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch backport/24511-to-camel-4.18.x in repository https://gitbox.apache.org/repos/asf/camel.git
commit 935c48ad4e330b4706bb1cccfafa15342af22dcf 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_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
