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 4f879148b118 CAMEL-25701: camel-azure-servicebus - renew a message
lock before the run that would come too late (#27986)
4f879148b118 is described below
commit 4f879148b118a131e3aa72467f26b02b6db532b9
Author: allthingssecurity <[email protected]>
AuthorDate: Sat Oct 17 14:43:25 2026 +0530
CAMEL-25701: camel-azure-servicebus - renew a message lock before the run
that would come too late (#27986)
The Camel lock renewer (CAMEL-23937) runs every 10 or 11 seconds (a 10000
ms period on the 1-second
timer of the period task scheduler) and renewed a lock only when it expired
within 10 seconds. A lock
that expired between 10 and 11 seconds after a run was left alone by that
run and reached by the next
one only after it had expired, or too close to the expiry for the renewal
call: the lock was lost and
Service Bus delivered the message again while the first exchange was still
being processed.
The renewer now renews a lock that expires within two intervals, so every
lock is renewed by a run that
comes at least about 9 seconds before it expires.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../azure/servicebus/ServiceBusConsumer.java | 6 ++-
.../azure/servicebus/ServiceBusConsumerTest.java | 47 ++++++++++++++++++++++
2 files changed, 51 insertions(+), 2 deletions(-)
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 ca9d65d9f710..c2c5a96c4949 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
@@ -342,9 +342,11 @@ public class ServiceBusConsumer extends DefaultConsumer
implements ShutdownAware
continue;
}
- // renew if lock is approaching expiry
+ // renew if the lock expires before the run after the next
one: the period task scheduler runs this
+ // task every LOCK_RENEW_INTERVAL_SECONDS or a second later,
so waiting for the next run when the lock
+ // expires within one interval can be too late, and the
renewal itself is a call to Service Bus
Instant lockExpiry = entry.lockedUntil.toInstant();
- if
(now.plusSeconds(LOCK_RENEW_INTERVAL_SECONDS).isAfter(lockExpiry)) {
+ if (now.plusSeconds(2L *
LOCK_RENEW_INTERVAL_SECONDS).isAfter(lockExpiry)) {
client.renewMessageLock(entry.message)
.subscribe(
newLockedUntil -> {
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 d490e0ff2e67..863134b0f81e 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
@@ -43,6 +43,7 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.EnumSource;
import org.mockito.ArgumentCaptor;
+import reactor.core.publisher.Mono;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
@@ -697,6 +698,52 @@ public class ServiceBusConsumerTest {
}
}
+ @Test
+ void lockRenewerRenewsALockThatWouldExpireBeforeItsNextRun() throws
Exception {
+
when(configuration.getServiceBusReceiveMode()).thenReturn(ServiceBusReceiveMode.PEEK_LOCK);
+ when(configuration.getMaxAutoLockRenewDuration()).thenReturn(300000L);
+ when(messageContext.getMessage()).thenReturn(message);
+
when(renewalClient.renewMessageLock(any(ServiceBusReceivedMessage.class)))
+ .thenReturn(Mono.just(OffsetDateTime.now().plusMinutes(1)));
+ try (ServiceBusConsumer consumer = new ServiceBusConsumer(endpoint,
processor)) {
+ consumer.doStart();
+
+ ArgumentCaptor<Runnable> taskCaptor = ArgumentCaptor.captor();
+ ArgumentCaptor<Long> periodCaptor = ArgumentCaptor.captor();
+
verify(periodTaskScheduler).schedulePeriodTask(taskCaptor.capture(),
periodCaptor.capture());
+ long periodMillis = periodCaptor.getValue();
+
+ // the lock expires 2 seconds after the earliest moment the
scheduler runs the renewer again
+
when(message.getLockedUntil()).thenReturn(OffsetDateTime.now().plus(Duration.ofMillis(periodMillis
+ 2000)));
+ processMessageCaptor.getValue().accept(messageContext);
+
+ taskCaptor.getValue().run();
+
+ // the next run comes at least periodMillis later (the period task
scheduler ticks once per second), the
+ // renewal itself is a call to Service Bus: waiting for that run
leaves at most 2 seconds, or none at all
+ verify(renewalClient).renewMessageLock(message);
+ }
+ }
+
+ @Test
+ void lockRenewerDoesNotRenewALockThatOutlastsTheNextRuns() 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();
+
+ ArgumentCaptor<Runnable> taskCaptor = ArgumentCaptor.captor();
+
verify(periodTaskScheduler).schedulePeriodTask(taskCaptor.capture(), anyLong());
+ processMessageCaptor.getValue().accept(messageContext);
+
+ taskCaptor.getValue().run();
+
+ verify(renewalClient,
never()).renewMessageLock(any(ServiceBusReceivedMessage.class));
+ }
+ }
+
@Test
void doStopCleansUpLockRenewer() throws Exception {
when(configuration.getServiceBusReceiveMode()).thenReturn(ServiceBusReceiveMode.PEEK_LOCK);
