From 25d484b45b6135678c08a841b364fdecbc29a214 Mon Sep 17 00:00:00 2001 From: Vijaya Gopal Yarramneni Date: Mon, 21 Dec 2020 21:27:21 -0800 Subject: [PATCH 1/4] Changing the message return daemon sleep to 10 milliseconds in CoreMessageReceiver. --- .../azure/servicebus/primitives/CoreMessageReceiver.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/CoreMessageReceiver.java b/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/CoreMessageReceiver.java index d3cb55f12559a..ff98d196fc461 100644 --- a/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/CoreMessageReceiver.java +++ b/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/CoreMessageReceiver.java @@ -66,7 +66,7 @@ public class CoreMessageReceiver extends ClientEntity implements IAmqpReceiver, IErrorContextProvider { private static final Logger TRACE_LOGGER = LoggerFactory.getLogger(CoreMessageReceiver.class); private static final Duration LINK_REOPEN_TIMEOUT = Duration.ofMinutes(5); // service closes link long before this timeout expires - private static final Duration RETURN_MESSAGES_DAEMON_WAKE_UP_INTERVAL = Duration.ofMillis(1); // Wakes up every 1 millisecond + private static final Duration RETURN_MESSAGES_DAEMON_WAKE_UP_INTERVAL = Duration.ofMillis(10); // Wakes up every few milliseconds private static final Duration UPDATE_STATE_REQUESTS_DAEMON_WAKE_UP_INTERVAL = Duration.ofMillis(500); // Wakes up every 500 milliseconds private static final Duration ZERO_TIMEOUT_APPROXIMATION = Duration.ofMillis(200); private static final int CREDIT_FLOW_BATCH_SIZE = 50; // Arbitrarily chosen 50 to avoid sending too many flows in case prefetch count is large From c53fb6840337b74ad0760180c886b986fb498844 Mon Sep 17 00:00:00 2001 From: Vijaya Gopal Yarramneni Date: Wed, 23 Dec 2020 22:39:35 -0800 Subject: [PATCH 2/4] Fixing a bug in sessino pump that is causing the pump to stop accepting sessions if a session lock is lost. --- .../servicebus/MessageAndSessionPump.java | 68 ++++++++++++++++--- .../azure/servicebus/ClientSessionTests.java | 6 ++ .../MessageAndSessionPumpTests.java | 28 ++++++++ 3 files changed, 94 insertions(+), 8 deletions(-) diff --git a/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/MessageAndSessionPump.java b/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/MessageAndSessionPump.java index 6604c793e4fdd..74eadb405736a 100644 --- a/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/MessageAndSessionPump.java +++ b/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/MessageAndSessionPump.java @@ -146,7 +146,7 @@ private synchronized void setHandlerRegistered() { private void receiveAndPumpMessage() { if (!this.getIsClosingOrClosed()) { - CompletableFuture receiveMessageFuture = this.innerReceiver.receiveAsync(this.messageHandlerOptions.getMessageWaitDuration()); + CompletableFuture receiveMessageFuture = receiveAsyncWrapper(this.innerReceiver, this.messageHandlerOptions.getMessageWaitDuration()); receiveMessageFuture.handleAsync((message, receiveEx) -> { if (receiveEx != null) { receiveEx = ExceptionUtil.extractAsyncCompletionCause(receiveEx); @@ -203,7 +203,7 @@ private void receiveAndPumpMessage() { dispositionPhase = ExceptionPhase.COMPLETE; if (this.messageHandlerOptions.isAutoComplete()) { TRACE_LOGGER.debug("Completing message with sequence number '{}'", message.getSequenceNumber()); - updateDispositionFuture = this.innerReceiver.completeAsync(message.getLockToken()); + updateDispositionFuture = completeAsyncWrapper(this.innerReceiver, message.getLockToken()); } else { updateDispositionFuture = CompletableFuture.completedFuture(null); } @@ -212,7 +212,7 @@ private void receiveAndPumpMessage() { dispositionPhase = ExceptionPhase.ABANDON; if (this.messageHandlerOptions.isAutoComplete()) { TRACE_LOGGER.debug("Abandoning message with sequence number '{}'", message.getSequenceNumber()); - updateDispositionFuture = this.innerReceiver.abandonAsync(message.getLockToken()); + updateDispositionFuture = abandonAsyncWrapper(this.innerReceiver, message.getLockToken()); } else { updateDispositionFuture = CompletableFuture.completedFuture(null); } @@ -290,7 +290,7 @@ private void acceptSessionAndPumpMessages() { private void receiveFromSessionAndPumpMessage(SessionTracker sessionTracker) { if (!this.getIsClosingOrClosed()) { IMessageSession session = sessionTracker.getSession(); - CompletableFuture receiverFuture = session.receiveAsync(this.sessionHandlerOptions.getMessageWaitDuration()); + CompletableFuture receiverFuture = receiveAsyncWrapper(session, this.sessionHandlerOptions.getMessageWaitDuration()); receiverFuture.handleAsync((message, receiveEx) -> { if (receiveEx != null) { receiveEx = ExceptionUtil.extractAsyncCompletionCause(receiveEx); @@ -350,7 +350,7 @@ private void receiveFromSessionAndPumpMessage(SessionTracker sessionTracker) { dispositionPhase = ExceptionPhase.COMPLETE; if (this.sessionHandlerOptions.isAutoComplete()) { TRACE_LOGGER.debug("Completing message with sequence number '{}'", message.getSequenceNumber()); - updateDispositionFuture = session.completeAsync(message.getLockToken()); + updateDispositionFuture = completeAsyncWrapper(session, message.getLockToken()); } else { updateDispositionFuture = CompletableFuture.completedFuture(null); } @@ -359,7 +359,7 @@ private void receiveFromSessionAndPumpMessage(SessionTracker sessionTracker) { dispositionPhase = ExceptionPhase.ABANDON; if (this.sessionHandlerOptions.isAutoComplete()) { TRACE_LOGGER.debug("Abandoning message with sequence number '{}'", message.getSequenceNumber()); - updateDispositionFuture = session.abandonAsync(message.getLockToken()); + updateDispositionFuture = abandonAsyncWrapper(session, message.getLockToken()); } else { updateDispositionFuture = CompletableFuture.completedFuture(null); } @@ -568,7 +568,7 @@ protected void loop() { if (renewInterval != null && !renewInterval.isNegative()) { this.timerFuture = Timer.schedule(() -> { TRACE_LOGGER.debug("Renewing lock on '{}'", this.messageIdentifier); - this.innerReceiver.renewMessageLockAsync(message).handleAsync((v, renewLockEx) -> { + renewMessageLockAsyncWrapper(this.innerReceiver, message).handleAsync((v, renewLockEx) -> { if (renewLockEx != null) { renewLockEx = ExceptionUtil.extractAsyncCompletionCause(renewLockEx); TRACE_LOGGER.info("Renewing lock on '{}' failed", this.messageIdentifier, renewLockEx); @@ -622,7 +622,7 @@ protected void loop() { if (renewInterval != null && !renewInterval.isNegative()) { this.timerFuture = Timer.schedule(() -> { TRACE_LOGGER.debug("Renewing lock on '{}'", this.sessionIdentifier); - this.session.renewSessionLockAsync().handleAsync((v, renewLockEx) -> { + renewSessionLockAsyncWrapper(this.session).handleAsync((v, renewLockEx) -> { if (renewLockEx != null) { renewLockEx = ExceptionUtil.extractAsyncCompletionCause(renewLockEx); TRACE_LOGGER.info("Renewing lock on '{}' failed", this.sessionIdentifier, renewLockEx); @@ -839,6 +839,58 @@ private void notifyExceptionToMessageHandler(Throwable ex, ExceptionPhase phase) this.customCodeExecutor.execute(() -> this.messageHandler.notifyException(ex, phase)); } } + + // These wrappers catch any synchronous exceptions and properly complete completablefutures with those excetions. + // Callers of these methods don't expect any synchronous exceptions. + private static CompletableFuture receiveAsyncWrapper(IMessageReceiver receiver, Duration serverWaitTime) { + try { + return receiver.receiveAsync(serverWaitTime); + } catch (Throwable t) { + CompletableFuture excetionalFuture = new CompletableFuture(); + excetionalFuture.completeExceptionally(t); + return excetionalFuture; + } + } + + private static CompletableFuture completeAsyncWrapper(IMessageReceiver receiver, UUID lockToken) { + try { + return receiver.completeAsync(lockToken); + } catch (Throwable t) { + CompletableFuture excetionalFuture = new CompletableFuture(); + excetionalFuture.completeExceptionally(t); + return excetionalFuture; + } + } + + private static CompletableFuture abandonAsyncWrapper(IMessageReceiver receiver, UUID lockToken) { + try { + return receiver.abandonAsync(lockToken); + } catch (Throwable t) { + CompletableFuture excetionalFuture = new CompletableFuture(); + excetionalFuture.completeExceptionally(t); + return excetionalFuture; + } + } + + private static CompletableFuture renewMessageLockAsyncWrapper(IMessageReceiver receiver, IMessage message) { + try { + return receiver.renewMessageLockAsync(message); + } catch (Throwable t) { + CompletableFuture excetionalFuture = new CompletableFuture(); + excetionalFuture.completeExceptionally(t); + return excetionalFuture; + } + } + + private static CompletableFuture renewSessionLockAsyncWrapper(IMessageSession session) { + try { + return session.renewSessionLockAsync(); + } catch (Throwable t) { + CompletableFuture excetionalFuture = new CompletableFuture(); + excetionalFuture.completeExceptionally(t); + return excetionalFuture; + } + } @Override public int getPrefetchCount() { diff --git a/sdk/servicebus/microsoft-azure-servicebus/src/test/java/com/microsoft/azure/servicebus/ClientSessionTests.java b/sdk/servicebus/microsoft-azure-servicebus/src/test/java/com/microsoft/azure/servicebus/ClientSessionTests.java index 7ed6de11467e8..6079e0fa44ff0 100644 --- a/sdk/servicebus/microsoft-azure-servicebus/src/test/java/com/microsoft/azure/servicebus/ClientSessionTests.java +++ b/sdk/servicebus/microsoft-azure-servicebus/src/test/java/com/microsoft/azure/servicebus/ClientSessionTests.java @@ -161,4 +161,10 @@ public void testSessionPumpRenewLock() throws InterruptedException, ServiceBusEx this.createClients(ReceiveMode.PEEKLOCK); MessageAndSessionPumpTests.testSessionPumpRenewLock(this.sendClient, this.receiveClient); } + + @Test + public void testSessionPumpLockLost() throws InterruptedException, ServiceBusException { + this.createClients(ReceiveMode.PEEKLOCK); + MessageAndSessionPumpTests.testSessionPumpLockLost(this.sendClient, this.receiveClient); + } } diff --git a/sdk/servicebus/microsoft-azure-servicebus/src/test/java/com/microsoft/azure/servicebus/MessageAndSessionPumpTests.java b/sdk/servicebus/microsoft-azure-servicebus/src/test/java/com/microsoft/azure/servicebus/MessageAndSessionPumpTests.java index 956fc3f2f9068..914e08e9ce9b6 100644 --- a/sdk/servicebus/microsoft-azure-servicebus/src/test/java/com/microsoft/azure/servicebus/MessageAndSessionPumpTests.java +++ b/sdk/servicebus/microsoft-azure-servicebus/src/test/java/com/microsoft/azure/servicebus/MessageAndSessionPumpTests.java @@ -255,6 +255,34 @@ public static void testSessionPumpRenewLock(IMessageSender sender, IMessageAndSe // So completes will pass before links are closed by teardown Thread.sleep(1000); } + + public static void testSessionPumpLockLost(IMessageSender sender, IMessageAndSessionPump sessionPump) throws InterruptedException, ServiceBusException { + int numSessions = 5; + int numMessagePerSession = 2; + ArrayList sessionIds = new ArrayList<>(); + for (int i = 0; i < numSessions; i++) { + String sessionId = StringUtil.getRandomString(); + sessionIds.add(sessionId); + for (int j = 0; j < numMessagePerSession; j++) { + Message message = new Message("AMQPMessage"); + message.setSessionId(sessionId); + sender.send(message); + } + } + + boolean autoComplete = true; + int sleepMinutes = 2; // This should be less than message lock duration of the queue or subscription + CountingSessionHandler sessionHandler = new CountingSessionHandler(sessionPump, !autoComplete, numSessions * numMessagePerSession, false, Duration.ofMinutes(sleepMinutes)); + // Register a handler that doesn't renew session lock + sessionPump.registerSessionHandler(sessionHandler, new SessionHandlerOptions(1, 1, autoComplete, Duration.ZERO), EXECUTOR_SERVICE); + // Sleep for a little longer than the handler sleep time. + // Session lock should be lost and a new session should be accepted. + Thread.sleep(1000 * 60 * 3); + + Assert.assertTrue("Another session not received by session pump, after session lock lost", sessionHandler.getReceivedSessions().size() > 1); + // So completes will pass before links are closed by teardown + Thread.sleep(1000); + } private static class CountingMessageHandler extends TestMessageHandler { private IMessageAndSessionPump messagePump; From 16b074b6ba543d7ccc63efd383b8aef923fc4997 Mon Sep 17 00:00:00 2001 From: Vijaya Gopal Yarramneni Date: Wed, 23 Dec 2020 22:50:16 -0800 Subject: [PATCH 3/4] Updating service bus track1 SDK version to 3.6.0. --- eng/spotbugs-aggregate-report/pom.xml | 2 +- eng/versioning/version_data.txt | 2 +- sdk/servicebus/microsoft-azure-servicebus/README.md | 2 +- sdk/servicebus/microsoft-azure-servicebus/pom.xml | 2 +- 4 files changed, 4 insertions(+), 4 deletions(-) diff --git a/eng/spotbugs-aggregate-report/pom.xml b/eng/spotbugs-aggregate-report/pom.xml index 0c100ad2014c8..7a2b72b62ce21 100644 --- a/eng/spotbugs-aggregate-report/pom.xml +++ b/eng/spotbugs-aggregate-report/pom.xml @@ -140,7 +140,7 @@ com.microsoft.azure azure-servicebus - 3.6.0-beta.1 + 3.6.0 diff --git a/eng/versioning/version_data.txt b/eng/versioning/version_data.txt index da53965c2792b..e7b147f2e7bc2 100644 --- a/eng/versioning/version_data.txt +++ b/eng/versioning/version_data.txt @@ -34,7 +34,7 @@ com.microsoft.azure:azure-keyvault-cryptography;1.2.4;1.3.0-beta.1 com.microsoft.azure:azure-keyvault-extensions;1.2.4;1.3.0-beta.1 com.microsoft.azure:azure-keyvault-test;1.2.3;1.2.4 com.microsoft.azure:azure-keyvault-webkey;1.2.4;1.3.0-beta.1 -com.microsoft.azure:azure-servicebus;3.5.1;3.6.0-beta.1 +com.microsoft.azure:azure-servicebus;3.5.1;3.6.0 com.microsoft.azure:azure-storage-blob;11.0.2;11.0.2 com.microsoft.azure.msi_auth_token_provider:azure-authentication-msi-token-provider;1.1.0-beta.1;1.1.0-beta.1 com.microsoft.azure:azure-eventgrid;1.4.0-beta.1;1.4.0-beta.1 diff --git a/sdk/servicebus/microsoft-azure-servicebus/README.md b/sdk/servicebus/microsoft-azure-servicebus/README.md index 35f0342a8562d..816bf1c1f6d6f 100644 --- a/sdk/servicebus/microsoft-azure-servicebus/README.md +++ b/sdk/servicebus/microsoft-azure-servicebus/README.md @@ -19,7 +19,7 @@ The package can be downloaded from [Maven](https://search.maven.org/artifact/com com.microsoft.azure azure-servicebus - 3.5.1 + 3.6.0 ``` [//]: # ({x-version-update-end}) diff --git a/sdk/servicebus/microsoft-azure-servicebus/pom.xml b/sdk/servicebus/microsoft-azure-servicebus/pom.xml index bcbd34e1a115f..e83f978a78588 100644 --- a/sdk/servicebus/microsoft-azure-servicebus/pom.xml +++ b/sdk/servicebus/microsoft-azure-servicebus/pom.xml @@ -6,7 +6,7 @@ 4.0.0 com.microsoft.azure azure-servicebus - 3.6.0-beta.1 + 3.6.0 Microsoft Azure SDK for Service Bus Java library for Azure Service Bus From 3c1cb4befa1421c4701596aba040c932dee8e94e Mon Sep 17 00:00:00 2001 From: Vijaya Gopal Yarramneni Date: Mon, 28 Dec 2020 15:23:00 -0800 Subject: [PATCH 4/4] Correcting a spelling. --- .../servicebus/MessageAndSessionPump.java | 30 +++++++++---------- 1 file changed, 15 insertions(+), 15 deletions(-) diff --git a/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/MessageAndSessionPump.java b/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/MessageAndSessionPump.java index 74eadb405736a..1e8922096eb10 100644 --- a/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/MessageAndSessionPump.java +++ b/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/MessageAndSessionPump.java @@ -846,9 +846,9 @@ private static CompletableFuture receiveAsyncWrapper(IMessageReceiver try { return receiver.receiveAsync(serverWaitTime); } catch (Throwable t) { - CompletableFuture excetionalFuture = new CompletableFuture(); - excetionalFuture.completeExceptionally(t); - return excetionalFuture; + CompletableFuture exceptionalFuture = new CompletableFuture(); + exceptionalFuture.completeExceptionally(t); + return exceptionalFuture; } } @@ -856,9 +856,9 @@ private static CompletableFuture completeAsyncWrapper(IMessageReceiver rec try { return receiver.completeAsync(lockToken); } catch (Throwable t) { - CompletableFuture excetionalFuture = new CompletableFuture(); - excetionalFuture.completeExceptionally(t); - return excetionalFuture; + CompletableFuture exceptionalFuture = new CompletableFuture(); + exceptionalFuture.completeExceptionally(t); + return exceptionalFuture; } } @@ -866,9 +866,9 @@ private static CompletableFuture abandonAsyncWrapper(IMessageReceiver rece try { return receiver.abandonAsync(lockToken); } catch (Throwable t) { - CompletableFuture excetionalFuture = new CompletableFuture(); - excetionalFuture.completeExceptionally(t); - return excetionalFuture; + CompletableFuture exceptionalFuture = new CompletableFuture(); + exceptionalFuture.completeExceptionally(t); + return exceptionalFuture; } } @@ -876,9 +876,9 @@ private static CompletableFuture renewMessageLockAsyncWrapper(IMessageR try { return receiver.renewMessageLockAsync(message); } catch (Throwable t) { - CompletableFuture excetionalFuture = new CompletableFuture(); - excetionalFuture.completeExceptionally(t); - return excetionalFuture; + CompletableFuture exceptionalFuture = new CompletableFuture(); + exceptionalFuture.completeExceptionally(t); + return exceptionalFuture; } } @@ -886,9 +886,9 @@ private static CompletableFuture renewSessionLockAsyncWrapper(IMessageSess try { return session.renewSessionLockAsync(); } catch (Throwable t) { - CompletableFuture excetionalFuture = new CompletableFuture(); - excetionalFuture.completeExceptionally(t); - return excetionalFuture; + CompletableFuture exceptionalFuture = new CompletableFuture(); + exceptionalFuture.completeExceptionally(t); + return exceptionalFuture; } }