[Service Bus] Bound sync acceptNextSession hang and add session-acquire retry backoff (#49093)

sdk/servicebus/azure-messaging-servicebus/CHANGELOG.md

@@ -8,6 +8,17 @@
 
 ### Bugs Fixed
 
+- Fixed `ServiceBusSessionReceiverClient.acceptNextSession()`/`acceptSession()` blocking for the full
+  operation timeout (~245s with default retry options) and throwing
+  `IllegalStateException: Timeout on blocking read ... (client-timeout)` when the broker accepts the
+  session-acquire link but never responds (a hung acquire). The session acquirer now bounds a single
+  acquire attempt on the synchronous (non-retry) path with a client-side guard of twice the
+  `tryTimeout`, disposing the half-open receive link when the guard fires so the broker-side session
+  lock is released instead of orphaned. On the retry-enabled path (session `ServiceBusProcessorClient`
+  and `ServiceBusSessionReceiverAsyncClient`), retries are now spaced by a bounded backoff instead of
+  retrying with no delay, preventing a tight CPU-burning loop when acquire attempts fail fast.
+  ([#49093](https://github.com/Azure/azure-sdk-for-java/issues/49093))
+
 ### Other Changes
 
 ## 7.17.19 (2026-07-01)

sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusSessionAcquirer.java

@@ -18,6 +18,7 @@
 import com.azure.messaging.servicebus.models.ServiceBusReceiveMode;
 import reactor.core.Disposable;
 import reactor.core.publisher.Mono;
+import reactor.core.publisher.SignalType;
 import reactor.core.scheduler.Schedulers;
 import reactor.util.retry.Retry;
 
@@ -35,8 +36,12 @@
  * <p>
  * The {@code timeoutRetryDisabled} is true when the session acquirer is used for synchronous {@link ServiceBusSessionReceiverClient}.
  * This allows the synchronous 'acceptNextSession()' API to propagate the broker timeout error if no session is available.
- * The 'acceptNextSession()' has a client-side timeout that is set slightly longer than the broker's timeout, ensuring
- * the broker's timeout usually triggers first (the client-side timeout still helps in case of unexpected hanging).
+ * On this non-retry path the acquirer bounds a single acquire attempt with a client-side guard set to twice the
+ * broker try-timeout, so the broker's own timeout usually triggers first (and its error propagates), while a hung
+ * acquire where the broker never responds still fails in bounded time instead of blocking the caller for the full
+ * operation timeout. If this client-side guard expires, the acquirer disposes the half-open receive link so the
+ * broker-side session lock (if any) is released rather than left orphaned. See
+ * <a href="https://github.com/Azure/azure-sdk-for-java/issues/49093">issue 49093</a>.
  * For ServiceBusSessionReceiverClient, if the library retries session-acquire on broker timeout, the client-side sync
  * timeout might expire while waiting. When client-side timeout expires like this, library cannot cancel the outstanding
  * acquire request to the broker, which means, the broker may still lock a session for an acquire request that nobody
@@ -45,7 +50,8 @@
  * </p>
  * <p>
  * For session enabled {@link ServiceBusProcessorClient} and {@link ServiceBusSessionReceiverAsyncClient},
- * the {@code timeoutRetryDisabled} is false, hence session acquirer retries on broker timeout.
+ * the {@code timeoutRetryDisabled} is false, hence session acquirer retries on broker timeout. Retries are spaced by a
+ * bounded backoff to avoid a tight CPU-burning loop when acquire attempts fail fast.
  * </p>
  */
 final class ServiceBusSessionAcquirer {
@@ -56,6 +62,10 @@ final class ServiceBusSessionAcquirer {
     private final MessagingEntityType entityType;
     private final Duration tryTimeout;
     private final boolean timeoutRetryDisabled;
+    // Client-side guard for the non-retry (sync ServiceBusSessionReceiverClient) path.
+    private final Duration clientSideTimeout;
+    // Backoff between session-acquire retries on the retry-enabled (async receiver / processor) path.
+    private final Duration retryBackoff;
     private final ServiceBusReceiveMode receiveMode;
     private final ConnectionCacheWrapper connectionCacheWrapper;
     private final Mono<ServiceBusManagementNode> sessionManagement;
@@ -82,6 +92,22 @@ final class ServiceBusSessionAcquirer {
         this.entityType = entityType;
         this.tryTimeout = tryTimeout;
         this.timeoutRetryDisabled = timeoutRetryDisabled;
+        // Bound a single acquire attempt on the sync (non-retry) path so that a hung broker/link that
+        // never sends its detach surfaces a timeout in ~2x the tryTimeout, instead of blocking the
+        // caller for the full operation timeout (the sum of all retry try-timeouts, ~245s with default
+        // AmqpRetryOptions). The 2x margin keeps the broker's own timeout detach the usual trigger, so
+        // the broker-timeout error still propagates in the healthy "no session available" case.
+        // https://github.com/Azure/azure-sdk-for-java/issues/49093
+        this.clientSideTimeout = tryTimeout.multipliedBy(2);
+        // Backoff between session-acquire retries to avoid a tight, CPU-burning retry loop when acquire
+        // attempts fail fast (e.g., the broker repeatedly detaches the accept link quickly). The previous
+        // Mono.delay(Duration.ZERO) ignored the configured retry delay and always spun; this honors the
+        // customer's configured AmqpRetryOptions delay, falling back to 100ms only when no usable delay
+        // (null/zero/negative) is configured. https://github.com/Azure/azure-sdk-for-java/issues/49093
+        final Duration configuredDelay = connectionCacheWrapper.getRetryOptions().getDelay();
+        this.retryBackoff = (configuredDelay == null || configuredDelay.isZero() || configuredDelay.isNegative())
+            ? Duration.ofMillis(100)
+            : configuredDelay;
         this.receiveMode = receiveMode;
         this.connectionCacheWrapper = connectionCacheWrapper;
         this.sessionManagement = connectionCacheWrapper.getConnection()
@@ -125,14 +151,22 @@ Mono<Session> acquire(String sessionId) {
      */
     private Mono<Session> acquireIntern(String sessionId) {
         if (timeoutRetryDisabled) {
-            return acquireSession(sessionId).onErrorResume(t -> {
-                if (isBrokerTimeoutError(t)) {
-                    // map the broker timeout to application-friendly TimeoutException.
-                    final Throwable e = new TimeoutException("com.microsoft:timeout").initCause(t);
-                    return publishError(sessionId, e, false);
-                }
-                return publishError(sessionId, t, true);
-            });
+            return acquireSession(sessionId).timeout(clientSideTimeout,
+                Mono.error(() -> new TimeoutException("Timed out waiting to acquire a session; the broker did not "
+                    + "respond within " + clientSideTimeout + " (client-side guard).")))
+                .onErrorResume(t -> {
+                    if (isBrokerTimeoutError(t)) {
+                        // map the broker timeout to application-friendly TimeoutException.
+                        final Throwable e = new TimeoutException("com.microsoft:timeout").initCause(t);
+                        return publishError(sessionId, e, false);
+                    }
+                    if (t instanceof TimeoutException) {
+                        // Client-side guard fired because the broker never responded (a hung acquire).
+                        // Treat it like the broker timeout: a retriable "no session acquired in time".
+                        return publishError(sessionId, t, false);
+                    }
+                    return publishError(sessionId, t, true);
+                });
         } else {
             return acquireSession(sessionId).timeout(tryTimeout)
                 .retryWhen(Retry.from(signals -> signals.flatMap(signal -> {
@@ -142,8 +176,10 @@ private Mono<Session> acquireIntern(String sessionId) {
                             .addKeyValue(ENTITY_PATH_KEY, entityPath)
                             .addKeyValue("attempt", signal.totalRetriesInARow())
                             .log("Timeout while acquiring session '{}'.", sessionName(sessionId), t);
-                        // retry session acquire using Schedulers.parallel() and free the QPid thread.
-                        return Mono.delay(Duration.ZERO);
+                        // Retry session acquire after a bounded backoff. Mono.delay hops to
+                        // Schedulers.parallel(), freeing the QPid thread; the non-zero backoff prevents a
+                        // tight CPU-burning loop when acquire attempts fail fast.
+                        return Mono.delay(retryBackoff);
                     }
                     return publishError(sessionId, t, true);
                 })));
@@ -165,7 +201,18 @@ private Mono<Session> acquireSession(String sessionId) {
             return createLink.flatMap(link -> {
                 // ServiceBusReceiveLink::getSessionProperties() await for link to "ACTIVE" then reads its properties.
                 return link.getSessionProperties()
-                    .flatMap(sessionProperties -> Mono.just(new Session(link, sessionProperties, sessionManagement)));
+                    .flatMap(sessionProperties -> Mono.just(new Session(link, sessionProperties, sessionManagement)))
+                    // If the caller abandons the acquire before a session is established (e.g., the
+                    // client-side timeout guard on the sync path, or a try-timeout on the retry path,
+                    // cancels the subscription), dispose the half-open receive link so it isn't leaked
+                    // and any broker-side session lock is released. On success the link ownership
+                    // transfers to the Session, so dispose only on cancellation.
+                    // https://github.com/Azure/azure-sdk-for-java/issues/49093
+                    .doFinally(signalType -> {
+                        if (signalType == SignalType.CANCEL) {
+                            link.dispose();
+                        }
+                    });
             });
         });
     }

sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusSessionReceiverClient.java

@@ -150,14 +150,18 @@ public final class ServiceBusSessionReceiverClient implements AutoCloseable {
 
     /**
      * Acquires a session lock for the next available session and creates a {@link ServiceBusReceiverClient}
-     * to receive messages from the session. It will wait until a session is available if no one is available
-     * immediately.
+     * to receive messages from the session. If no session is available immediately, it waits for one; if none
+     * becomes available within the operation timeout it throws a retriable timeout error rather than blocking
+     * indefinitely, so callers typically invoke this in a loop.
      *
      * @return A {@link ServiceBusReceiverClient} that is tied to the available session.
      *
      * @throws UnsupportedOperationException if the queue or topic subscription is not session-enabled.
-     * @throws AmqpException if the operation times out. The timeout duration is the tryTimeout
-     *      of when you build this client with the {@link ServiceBusClientBuilder#retryOptions(AmqpRetryOptions)}.
+     * @throws IllegalStateException if no session becomes available within the operation timeout (the total
+     *      of all retry attempts derived from {@link ServiceBusClientBuilder#retryOptions(AmqpRetryOptions)});
+     *      this is retriable, so callers typically retry.
+     * @throws AmqpException if acquiring the session fails at the AMQP level (for example an authorization or
+     *      connection error surfaced by the underlying asynchronous accept).
      */
     @ServiceMethod(returns = ReturnType.SINGLE)
     public ServiceBusReceiverClient acceptNextSession() {
@@ -185,8 +189,12 @@ public ServiceBusReceiverClient acceptNextSession() {
      * @throws IllegalArgumentException if {@code sessionId} is empty.
      * @throws UnsupportedOperationException if the queue or topic subscription is not session-enabled.
      * @throws ServiceBusException if the lock cannot be acquired.
-     * @throws AmqpException if the operation times out. The timeout duration is the tryTimeout
-     *      of when you build this client with the {@link ServiceBusClientBuilder#retryOptions(AmqpRetryOptions)}.
+     * @throws IllegalStateException if the session is not acquired within the operation timeout (the total
+     *      of all retry attempts derived from {@link ServiceBusClientBuilder#retryOptions(AmqpRetryOptions)});
+     *      this is retriable, so callers typically retry.
+     * @throws AmqpException if acquiring the session fails at the AMQP level (for example the session is already
+     *      locked by another client, or an authorization or connection error surfaced by the underlying
+     *      asynchronous accept).
      */
     @ServiceMethod(returns = ReturnType.SINGLE)
     public ServiceBusReceiverClient acceptSession(String sessionId) {

sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusSessionAcquirerIsolatedTest.java

@@ -3,6 +3,7 @@
 
 package com.azure.messaging.servicebus;
 
+import com.azure.core.amqp.AmqpRetryOptions;
 import com.azure.core.amqp.exception.AmqpErrorCondition;
 import com.azure.core.amqp.exception.AmqpErrorContext;
 import com.azure.core.amqp.exception.AmqpException;
@@ -47,6 +48,7 @@ public class ServiceBusSessionAcquirerIsolatedTest {
     private static final MessagingEntityType ENTITY_TYPE = MessagingEntityType.QUEUE;
     private static final ServiceBusReceiveMode RECEIVE_MODE = ServiceBusReceiveMode.PEEK_LOCK;
     private static final Duration TRY_TIMEOUT = Duration.ofSeconds(5);
+    private static final Duration RETRY_BACKOFF = Duration.ofSeconds(2);
     private static final Duration AWAIT_DURATION = TRY_TIMEOUT.plusSeconds(5);
     private static final AmqpException BROKER_TIMEOUT_ERROR = new AmqpException(true, AmqpErrorCondition.TIMEOUT_ERROR,
         "com.microsoft:timeout", new AmqpErrorContext(ENTITY_PATH));
@@ -107,6 +109,56 @@ public void shouldPropagateAnyErrorIfRetryDisabled() {
         Assertions.assertEquals(0, onCreateSessionLink.pending());
     }
 
+    @Test
+    @Execution(ExecutionMode.SAME_THREAD)
+    public void shouldClientSideTimeoutAndDisposeLinkWhenAcquireHangsIfRetryDisabled() {
+        // A receive link whose getSessionProperties() never emits simulates a hung acquire where the
+        // broker accepts the link but never sends its detach. Without a client-side guard this would
+        // block the caller for the full operation timeout (issue #49093).
+        final ServiceBusReceiveLink hungLink = mock(ServiceBusReceiveLink.class);
+        when(hungLink.getSessionProperties()).thenReturn(Mono.never());
+        final Deque<Mono<ServiceBusReceiveLink>> sessionLinks = new ArrayDeque<>(1);
+        sessionLinks.add(Mono.just(hungLink));
+        final OnCreateSessionLink onCreateSessionLink = new OnCreateSessionLink(sessionLinks);
+        final ConnectionCacheWrapper cacheWrapper = createMockConnectionWrapper(onCreateSessionLink);
+        final ServiceBusSessionAcquirer sessionAcquirer = createSessionAcquirer(cacheWrapper, true);
+
+        // The client-side guard is 2 * tryTimeout; await slightly longer so it fires.
+        final Duration awaitPastGuard = TRY_TIMEOUT.multipliedBy(2).plusSeconds(1);
+        try (VirtualTimeStepVerifier verifier = new VirtualTimeStepVerifier()) {
+            verifier.create(sessionAcquirer::acquire).thenAwait(awaitPastGuard).verifyErrorSatisfies(e -> {
+                Assertions.assertInstanceOf(TimeoutException.class, e);
+            });
+        }
+        Assertions.assertEquals(0, onCreateSessionLink.pending());
+        // The half-open link must be disposed when the client-side guard cancels the acquire.
+        Mockito.verify(hungLink).dispose();
+    }
+
+    @Test
+    @Execution(ExecutionMode.SAME_THREAD)
+    public void shouldBackOffBetweenRetriesIfRetryEnabled() {
+        // First attempt fails with a (retriable) broker timeout, then a terminal error. The retry must
+        // wait for the backoff before the second acquire attempt, i.e. it must not spin (issue #49093).
+        final Deque<Mono<ServiceBusReceiveLink>> sessionLinks = new ArrayDeque<>(2);
+        sessionLinks.add(Mono.error(BROKER_TIMEOUT_ERROR));
+        final RuntimeException error = new RuntimeException();
+        sessionLinks.add(Mono.error(error));
+        final OnCreateSessionLink onCreateSessionLink = new OnCreateSessionLink(sessionLinks);
+        final ConnectionCacheWrapper cacheWrapper = createMockConnectionWrapper(onCreateSessionLink);
+        final ServiceBusSessionAcquirer sessionAcquirer = createSessionAcquirer(cacheWrapper, false);
+
+        try (VirtualTimeStepVerifier verifier = new VirtualTimeStepVerifier()) {
+            verifier.create(sessionAcquirer::acquire)
+                .thenAwait(RETRY_BACKOFF.dividedBy(2))
+                // Before the backoff elapses the second attempt must not have been made yet.
+                .then(() -> Assertions.assertEquals(1, onCreateSessionLink.pending()))
+                .thenAwait(RETRY_BACKOFF)
+                .verifyErrorSatisfies(e -> Assertions.assertEquals(error, e));
+        }
+        Assertions.assertEquals(0, onCreateSessionLink.pending());
+    }
+
     @Test
     @Execution(ExecutionMode.SAME_THREAD)
     public void shouldRetryOnBrokerTimeoutErrorIfRetryEnabled() {
@@ -155,6 +207,7 @@ private ConnectionCacheWrapper createMockConnectionWrapper(OnCreateSessionLink o
         final ConnectionCacheWrapper cacheWrapper = Mockito.mock(ConnectionCacheWrapper.class);
         when(cacheWrapper.isV2()).thenReturn(true);
         when(cacheWrapper.getConnection()).thenReturn(Mono.just(connection));
+        when(cacheWrapper.getRetryOptions()).thenReturn(new AmqpRetryOptions().setDelay(RETRY_BACKOFF));
         return cacheWrapper;
     }
 

sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusSessionReceiverClientTest.java

@@ -5,6 +5,8 @@
 
 import com.azure.messaging.servicebus.implementation.instrumentation.ReceiverKind;
 import com.azure.messaging.servicebus.implementation.instrumentation.ServiceBusReceiverInstrumentation;
+import com.azure.core.amqp.exception.AmqpErrorContext;
+import com.azure.core.amqp.exception.AmqpException;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -15,8 +17,11 @@
 import reactor.core.publisher.Mono;
 
 import java.time.Duration;
+import java.util.concurrent.TimeoutException;
 
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.Mockito.when;
@@ -78,4 +83,31 @@ void acceptNextSessionTimeout() {
 
         assertThrows(IllegalStateException.class, () -> sessionClient.acceptNextSession());
     }
+
+    @Test
+    void acceptNextSessionTimeoutIsIllegalStateWithTimeoutCause() {
+        when(sessionAsyncClient.acceptNextSession())
+            .thenReturn(Mono.just(asyncClient).delayElement(Duration.ofMillis(500)));
+        ServiceBusSessionReceiverClient sessionClient
+            = new ServiceBusSessionReceiverClient(sessionAsyncClient, false, Duration.ofMillis(50));
+
+        // The operation timeout is surfaced as IllegalStateException (NOT AmqpException), caused by a
+        // TimeoutException. This backs the acceptNextSession() Javadoc @throws contract.
+        final IllegalStateException ex
+            = assertThrows(IllegalStateException.class, () -> sessionClient.acceptNextSession());
+        assertInstanceOf(TimeoutException.class, ex.getCause());
+    }
+
+    @Test
+    void acceptNextSessionPropagatesAmqpException() {
+        // An AMQP-level failure from the underlying async accept propagates AS-IS (AmqpException), i.e.
+        // AmqpException is NOT the operation-timeout signal. This backs the corrected Javadoc @throws.
+        final AmqpException amqpError = new AmqpException(false, "AMQP-level failure", new AmqpErrorContext("fqdn"));
+        when(sessionAsyncClient.acceptNextSession()).thenReturn(Mono.error(amqpError));
+        ServiceBusSessionReceiverClient sessionClient
+            = new ServiceBusSessionReceiverClient(sessionAsyncClient, false, Duration.ofSeconds(5));
+
+        final AmqpException thrown = assertThrows(AmqpException.class, () -> sessionClient.acceptNextSession());
+        assertSame(amqpError, thrown);
+    }
 }