This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch fix/CAMEL-23937 in repository https://gitbox.apache.org/repos/asf/camel.git
commit 13392fd1401e7417383ad6265964ccab8bccfe63 Author: Claus Ibsen <[email protected]> AuthorDate: Wed Jul 8 09:09:52 2026 +0200 CAMEL-23937: Add Camel-managed lock renewal for Azure Service Bus async routes The Azure SDK's built-in lock renewal is tied to the processMessage callback duration, which returns immediately for async Camel routes. This makes maxAutoLockRenewDuration ineffective, causing silent message-lock expiry, duplicate delivery, and data loss. This adds a LockRenewer that uses a ServiceBusReceiverAsyncClient to periodically renew message locks for in-flight exchanges. It integrates with Camel's PeriodTaskScheduler for operational visibility and uses non-blocking Reactor Mono subscriptions for concurrent renewals. Session-enabled consumers are excluded (follow-up in CAMEL-23938). Co-Authored-By: Claude Opus 4.6 <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../azure/servicebus/ServiceBusConsumer.java | 122 +++++++++++++++++++++ .../servicebus/client/ServiceBusClientFactory.java | 28 +++++ .../azure/servicebus/ServiceBusConsumerTest.java | 102 +++++++++++++++++ 3 files changed, 252 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..3a5a4d18d03a 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); @@ -114,6 +147,14 @@ public class ServiceBusConsumer extends DefaultConsumer implements ShutdownAware @Override protected void doShutdown() throws Exception { + 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 +239,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..fb82e039f1f0 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,98 @@ 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); + when(message.getLockedUntil()).thenReturn(OffsetDateTime.now().plusMinutes(1)); + try (ServiceBusConsumer consumer = new ServiceBusConsumer(endpoint, processor)) { + consumer.doStart(); + + processMessageCaptor.getValue().accept(messageContext); + + // pendingExchanges tracks the in-flight message + assertThat(consumer.getPendingExchangesSize()).isEqualTo(1); + + // complete the exchange — should remove from lock renewer too + Exchange exchange = exchangeCaptor.getValue(); + Synchronization synchronization = exchange.getExchangeExtension().handoverCompletions().get(0); + synchronization.onComplete(exchange); + + assertThat(consumer.getPendingExchangesSize()).isZero(); + } + } + + @Test + void doShutdownCleansUpLockRenewer() 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.doShutdown(); + + verify(renewalClient).close(); + verify(client).close(); + } + @Test void customProcessorClient() throws Exception { ServiceBusProcessorClient customClient = mock();
