From dfc5453c7c729361f9aea28de4bf983309a002ff Mon Sep 17 00:00:00 2001 From: "Karen Salazar (from Dev Box)" Date: Tue, 14 Jul 2026 10:01:39 -0700 Subject: [PATCH 1/2] [Service Bus] Bound sync acceptNextSession hang and add session-acquire retry backoff (#49093) Bounds a single synchronous session-acquire attempt with a client-side guard of 2x tryTimeout (disposing the half-open link when it fires) so a hung acceptNextSession()/acceptSession() fails in bounded time instead of blocking for the full ~245s operation timeout. Adds a bounded backoff between session-acquire retries on the session processor/async receiver path to avoid a CPU-burning tight loop when acquire attempts fail fast. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 406b56c0-6215-4747-b9bf-c7d7b8ab1e89 --- .../azure-messaging-servicebus/CHANGELOG.md | 11 +++ .../servicebus/ServiceBusSessionAcquirer.java | 74 +++++++++++++++---- .../ServiceBusSessionReceiverClient.java | 7 +- ...ServiceBusSessionAcquirerIsolatedTest.java | 53 +++++++++++++ 4 files changed, 129 insertions(+), 16 deletions(-) diff --git a/sdk/servicebus/azure-messaging-servicebus/CHANGELOG.md b/sdk/servicebus/azure-messaging-servicebus/CHANGELOG.md index ec97b08279fc..8aaa63c47759 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.18.0-beta.2 (2026-06-22) 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..d0960aa28abb 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,21 @@ 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). A zero + // delay was previously used only to hop off the QPid thread; the non-zero backoff below keeps + // that thread hop while preventing the busy loop. 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 +150,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 +175,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 +200,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..17d7baf5c42c 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,12 +150,15 @@ 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 IllegalStateException if no session becomes available within the operation timeout (derived from + * the {@link ServiceBusClientBuilder#retryOptions(AmqpRetryOptions)}); this is retriable. * @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)}. */ 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; } From c11773cc9d273d10269145a159aee3ae86cf2287 Mon Sep 17 00:00:00 2001 From: "Karen Salazar (from Dev Box)" Date: Wed, 15 Jul 2026 22:34:07 -0700 Subject: [PATCH 2/2] [Service Bus] Address PR review feedback for session-acquire fix (#49093) Clarify acceptNextSession()/acceptSession() @throws Javadoc: the operation timeout surfaces as IllegalStateException, while AmqpException covers AMQP-level failures (not the operation timeout). Add regression tests asserting both exception paths. Clarify the retry-backoff comment to state it honors the customer's configured AmqpRetryOptions delay (defaulting to 100ms only when no usable delay is configured); no behavior change. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 406b56c0-6215-4747-b9bf-c7d7b8ab1e89 --- .../servicebus/ServiceBusSessionAcquirer.java | 7 ++-- .../ServiceBusSessionReceiverClient.java | 17 ++++++---- .../ServiceBusSessionReceiverClientTest.java | 32 +++++++++++++++++++ 3 files changed, 47 insertions(+), 9 deletions(-) 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 d0960aa28abb..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 @@ -100,9 +100,10 @@ final class ServiceBusSessionAcquirer { // 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). A zero - // delay was previously used only to hop off the QPid thread; the non-zero backoff below keeps - // that thread hop while preventing the busy loop. https://github.com/Azure/azure-sdk-for-java/issues/49093 + // 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) 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 17d7baf5c42c..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 @@ -157,10 +157,11 @@ public final class ServiceBusSessionReceiverClient implements AutoCloseable { * @return A {@link ServiceBusReceiverClient} that is tied to the available session. * * @throws UnsupportedOperationException if the queue or topic subscription is not session-enabled. - * @throws IllegalStateException if no session becomes available within the operation timeout (derived from - * the {@link ServiceBusClientBuilder#retryOptions(AmqpRetryOptions)}); this is retriable. - * @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() { @@ -188,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/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); + } }