diff --git a/sdk/servicebus/azure-messaging-servicebus/CHANGELOG.md b/sdk/servicebus/azure-messaging-servicebus/CHANGELOG.md
index 7034ec765efe..0f959b16ea27 100644
--- a/sdk/servicebus/azure-messaging-servicebus/CHANGELOG.md
+++ b/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)
diff --git a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusSessionAcquirer.java b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusSessionAcquirer.java
index ee9a0a9e0399..ec8b8365ae7d 100644
--- a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusSessionAcquirer.java
+++ b/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 @@
*
* 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
+ * issue 49093.
* 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 @@
*
*
* 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.
*
*/
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 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 acquire(String sessionId) {
*/
private Mono 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 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 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();
+ }
+ });
});
});
}
diff --git a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusSessionReceiverClient.java b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusSessionReceiverClient.java
index 351d29db6d06..c8cc9cfff0c9 100644
--- a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusSessionReceiverClient.java
+++ b/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) {
diff --git a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusSessionAcquirerIsolatedTest.java b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusSessionAcquirerIsolatedTest.java
index eb0113ab82ca..321acb70c266 100644
--- a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusSessionAcquirerIsolatedTest.java
+++ b/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> 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> 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;
}
diff --git a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusSessionReceiverClientTest.java b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusSessionReceiverClientTest.java
index f8c285c69968..926cf7f7088b 100644
--- a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusSessionReceiverClientTest.java
+++ b/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);
+ }
}