Allow broker session acquire timeout to propagate to acceptSession sync api (Azure#42838)

* Allow broker session acquire timeout to propagate to acceptSession sync api

* Cleanup the java-doc, code comments

* cleanup code, and doc

* changelog, comment cleanup

* add feature entry

* readability improvements

* adding unit tests for ServiceBusSessionAcquirer

* Improve existing behaviour of throwing IllegalStateException by including cause as TimeoutException

* spot check, minor log tweak

Author: anuchandy

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

@@ -5,6 +5,7 @@
 ### Features Added
 
 - Enabled RequestResponseChannelCache (CBS, Management channel cache) and ReactorSessionCache by default. ([42641](https://github.com/Azure/azure-sdk-for-java/pull/42641))
+- Improves the synchronous `acceptNextSession` and `acceptSession` APIs of `ServiceBusSessionReceiverClient` to reduce the chances of the broker holding session lock for some time when client-side timeout occurs. ([42838](https://github.com/Azure/azure-sdk-for-java/pull/42838))
 
 ### Breaking Changes
 

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

@@ -2108,8 +2108,12 @@ SessionsMessagePump buildPumpForProcessor(ClientLogger logger,
             final ConnectionCacheWrapper connectionCacheWrapper = new ConnectionCacheWrapper(
                 getOrCreateConnectionCache(messageSerializer, meter, useSessionChannelCache));
 
-            final ServiceBusSessionAcquirer sessionAcquirer = new ServiceBusSessionAcquirer(logger, clientIdentifier,
-                entityPath, entityType, receiveMode, retryOptions.getTryTimeout(), connectionCacheWrapper);
+            // For session enabled ServiceBusProcessorClient, the session acquire should be retried if broker timeout
+            // due to no session (see the type ServiceBusSessionAcquirer).
+            final boolean timeoutRetryDisabled = false;
+            final ServiceBusSessionAcquirer sessionAcquirer
+                = new ServiceBusSessionAcquirer(logger, clientIdentifier, entityPath, entityType, receiveMode,
+                    retryOptions.getTryTimeout(), timeoutRetryDisabled, connectionCacheWrapper);
 
             final ServiceBusReceiverInstrumentation instrumentation = new ServiceBusReceiverInstrumentation(
                 createTracer(clientOptions), meter, connectionCacheWrapper.getFullyQualifiedNamespace(), entityPath,
@@ -2138,9 +2142,8 @@ SessionsMessagePump buildPumpForProcessor(ClientLogger logger,
          *     queueName()} or {@link #topicName(String) topicName()}, respectively.
          */
         public ServiceBusSessionReceiverAsyncClient buildAsyncClient() {
-            final boolean isSessionReactorReceiveOnV2
-                = v2StackSupport.isSessionReactorAsyncReceiveEnabled(configuration);
-            return buildAsyncClient(true, isSessionReactorReceiveOnV2);
+            final boolean isV2 = v2StackSupport.isSessionReactorAsyncReceiveEnabled(configuration);
+            return buildAsyncClient(false, isV2);
         }
 
         /**
@@ -2157,19 +2160,27 @@ public ServiceBusSessionReceiverAsyncClient buildAsyncClient() {
          *     queueName()} or {@link #topicName(String) topicName()}, respectively.
          */
         public ServiceBusSessionReceiverClient buildClient() {
-            final boolean isSessionSyncReceiveOnV2 = v2StackSupport.isSessionSyncReceiveEnabled(configuration);
+            final boolean isV2 = v2StackSupport.isSessionSyncReceiveEnabled(configuration);
             final boolean isPrefetchDisabled = prefetchCount == 0;
-            return new ServiceBusSessionReceiverClient(buildAsyncClient(false, isSessionSyncReceiveOnV2),
-                isPrefetchDisabled, MessageUtils.getTotalTimeout(retryOptions));
+            return new ServiceBusSessionReceiverClient(buildAsyncClient(true, isV2), isPrefetchDisabled,
+                MessageUtils.getTotalTimeout(retryOptions));
         }
 
-        // Common function to build Session-Enabled Receiver-Client - For Async[Reactor]Client Or to back SyncClient.
-        private ServiceBusSessionReceiverAsyncClient buildAsyncClient(boolean isAutoCompleteAllowed, boolean isV2) {
+        /**
+         * Common function to build a {@link ServiceBusSessionReceiverAsyncClient} which is either used directly
+         * as asynchronous client or to back a synchronous {@link ServiceBusSessionReceiverClient}.
+         *
+         * @param isForSyncMode {@code true} if the client is build to back synchronous client.
+         * @param isV2 whether V2 stack should be enabled.
+         *
+         * @return async client to obtain session from session enabled entity.
+         */
+        private ServiceBusSessionReceiverAsyncClient buildAsyncClient(boolean isForSyncMode, boolean isV2) {
             final MessagingEntityType entityType
                 = validateEntityPaths(connectionStringEntityName, topicName, queueName);
             final String entityPath = getEntityPath(entityType, queueName, topicName, subscriptionName, SubQueue.NONE);
 
-            if (!isAutoCompleteAllowed && enableAutoComplete) {
+            if (isForSyncMode && enableAutoComplete) {
                 LOGGER.warning(
                     "'enableAutoComplete' is not supported in synchronous client except through callback receive.");
                 enableAutoComplete = false;
@@ -2208,12 +2219,16 @@ private ServiceBusSessionReceiverAsyncClient buildAsyncClient(boolean isAutoComp
                 clientIdentifier = UUID.randomUUID().toString();
             }
 
+            // For ServiceBusSessionReceiverClient, the session acquire should not be retried if broker timeout due to
+            // no session (see the type ServiceBusSessionAcquirer), such timeout are propagated to the acceptNextSession() caller.
+            final boolean timeoutRetryDisabled = isV2 && isForSyncMode;
+
             final ServiceBusReceiverInstrumentation instrumentation = new ServiceBusReceiverInstrumentation(
                 createTracer(clientOptions), meter, connectionCacheWrapper.getFullyQualifiedNamespace(), entityPath,
                 subscriptionName, ReceiverKind.ASYNC_RECEIVER);
             return new ServiceBusSessionReceiverAsyncClient(connectionCacheWrapper.getFullyQualifiedNamespace(),
                 entityPath, entityType, receiverOptions, connectionCacheWrapper, instrumentation, messageSerializer,
-                onClientClose, clientIdentifier);
+                onClientClose, clientIdentifier, timeoutRetryDisabled);
         }
     }
 

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

@@ -3,9 +3,12 @@
 
 package com.azure.messaging.servicebus;
 
+import com.azure.core.amqp.AmqpClientOptions;
+import com.azure.core.amqp.AmqpRetryOptions;
 import com.azure.core.amqp.exception.AmqpErrorCondition;
 import com.azure.core.amqp.exception.AmqpException;
 import com.azure.core.amqp.implementation.StringUtil;
+import com.azure.core.amqp.implementation.handler.ReceiveLinkHandler2;
 import com.azure.core.util.logging.ClientLogger;
 import com.azure.messaging.servicebus.implementation.MessagingEntityType;
 import com.azure.messaging.servicebus.implementation.ServiceBusManagementNode;
@@ -26,26 +29,59 @@
 
 import static com.azure.core.amqp.implementation.ClientConstants.ENTITY_PATH_KEY;
 
+/**
+ * A type to acquire a session from a session enabled Service Bus entity. If the broker cannot find a session within
+ * the timeout, it returns a timeout error. The acquirer retries on timeout unless disabled via {@code timeoutRetryDisabled}.
+ * <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).
+ * 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
+ * is waiting on, resulting that session to be unavailable for any other 'acceptNextSession()' until initial broker
+ * lock expires. Hence, library propagate the broker timeout error in ServiceBusSessionReceiverClient case.
+ * </p>
+ * <p>
+ * For session enabled {@link ServiceBusProcessorClient} and {@link ServiceBusSessionReceiverAsyncClient},
+ * the {@code timeoutRetryDisabled} is false, hence session acquirer retries on broker timeout.
+ * </p>
+ */
 final class ServiceBusSessionAcquirer {
     private static final String TRACKING_ID_KEY = "trackingId";
     private final ClientLogger logger;
     private final String identifier;
     private final String entityPath;
     private final MessagingEntityType entityType;
-    private final Duration sessionActiveTimeout;
+    private final Duration tryTimeout;
+    private final boolean timeoutRetryDisabled;
     private final ServiceBusReceiveMode receiveMode;
     private final ConnectionCacheWrapper connectionCacheWrapper;
     private final Mono<ServiceBusManagementNode> sessionManagement;
 
+    /**
+     * Creates ServiceBusSessionAcquirer to acquire session from a session enabled entity.
+     *
+     * @param logger the logger to use.
+     * @param identifier the client identifier, currently callsites uses {@link AmqpClientOptions#getIdentifier()}.
+     * @param entityPath path to the session enabled entity.
+     * @param entityType the entity type (e.g., queue, topic subscription)
+     * @param receiveMode the mode of receiving messages from the acquired session.
+     * @param tryTimeout the try timeout, currently callsites uses {@link AmqpRetryOptions#getTryTimeout()}}.
+     * @param timeoutRetryDisabled if session acquire retry should be disabled when broker timeout on no session.
+     * @param connectionCacheWrapper the connection cache.
+     */
     ServiceBusSessionAcquirer(ClientLogger logger, String identifier, String entityPath, MessagingEntityType entityType,
-        ServiceBusReceiveMode receiveMode, Duration sessionActiveTimeout,
+        ServiceBusReceiveMode receiveMode, Duration tryTimeout, boolean timeoutRetryDisabled,
         ConnectionCacheWrapper connectionCacheWrapper) {
         assert connectionCacheWrapper.isV2();
         this.logger = logger;
         this.identifier = identifier;
         this.entityPath = entityPath;
         this.entityType = entityType;
-        this.sessionActiveTimeout = sessionActiveTimeout;
+        this.tryTimeout = tryTimeout;
+        this.timeoutRetryDisabled = timeoutRetryDisabled;
         this.receiveMode = receiveMode;
         this.connectionCacheWrapper = connectionCacheWrapper;
         this.sessionManagement = connectionCacheWrapper.getConnection()
@@ -78,53 +114,135 @@ Mono<Session> acquire(String sessionId) {
         return acquireIntern(sessionId);
     }
 
+    /**
+     * Tries to acquire a session from the broker by opening an AMQP receive link. When the acquire attempt timeout then
+     * the api will retry if {@code timeoutRetryDisabled} is set to {@code false}. In case an error needs to be propagated,
+     * api publishes the error using bounded-elastic Thread.
+     *
+     * @param sessionId the unique id of the specific session to acquire, a value {@code null} means acquire any free session.
+     * @return A Mono that completes with the acquired session, the Mono can emit {@link AmqpException} if the acquirer
+     * is already disposed or {@link TimeoutException} if session acquire timeouts and {@code timeoutRetryDisabled} set to true.
+     */
     private Mono<Session> acquireIntern(String sessionId) {
-        return Mono
-            .defer(() -> createSessionReceiveLink(sessionId).flatMap(sessionLink -> sessionLink.getSessionProperties() // Await for sessionLink to "ACTIVE" then reads its properties
-                .flatMap(sessionProperties -> {
-                    return Mono.just(new Session(sessionLink, sessionProperties, sessionManagement));
-                })))
-            .timeout(sessionActiveTimeout)
-            .retryWhen(Retry.from(retrySignals -> retrySignals.flatMap(signal -> {
-                final Throwable failure = signal.failure();
-                logger.atInfo()
-                    .addKeyValue(ENTITY_PATH_KEY, entityPath)
-                    .addKeyValue("attempt", signal.totalRetriesInARow())
-                    .log(sessionId == null
-                        ? "Error occurred while getting unnamed session."
-                        : "Error occurred while getting session " + sessionId, failure);
-
-                if (failure instanceof TimeoutException) {
-                    return Mono.delay(Duration.ZERO);
-                } else if (failure instanceof AmqpException
-                    && ((AmqpException) failure).getErrorCondition() == AmqpErrorCondition.TIMEOUT_ERROR) {
-                    // The link closed remotely with 'Detach {errorCondition:com.microsoft:timeout}' frame because
-                    // the broker waited for N seconds (60 sec hard limit today) but there was no free or new session.
-                    //
-                    // Given N seconds elapsed since the last session acquire attempt, request for a session on
-                    // the 'parallel' Scheduler and free the QPid thread for other IO.
-                    //
-                    return Mono.delay(Duration.ZERO);
-                } else {
-                    final long id = System.nanoTime();
-                    logger.atInfo().addKeyValue(TRACKING_ID_KEY, id).log("Unable to acquire a session.", failure);
-                    // The link-endpoint-state publisher will emit error on the QPid Thread, that is a non-blocking Thread,
-                    // publish the error on the (blockable) bounded-elastic thread to free QPid thread and to allow
-                    // any blocking operation that downstream may do.
-                    return Mono.<Long>error(failure)
-                        .publishOn(Schedulers.boundedElastic())
-                        .doOnError(e -> logger.atInfo()
-                            .addKeyValue(TRACKING_ID_KEY, id)
-                            .log("Emitting the error signal received for session acquire attempt.", e));
+        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);
+            });
+        } else {
+            return acquireSession(sessionId).timeout(tryTimeout)
+                .retryWhen(Retry.from(signals -> signals.flatMap(signal -> {
+                    final Throwable t = signal.failure();
+                    if (isTimeoutError(t)) {
+                        logger.atVerbose()
+                            .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);
+                    }
+                    return publishError(sessionId, t, true);
+                })));
+        }
+    }
+
+    /**
+     * Tries to acquire a session from the broker by opening an AMQP receive link.
+     *
+     * @param sessionId the unique id of the session to acquire, a value {@code null} means acquire any free session.
+     *
+     * @return the acquired session.
+     */
+    private Mono<Session> acquireSession(String sessionId) {
+        return Mono.defer(() -> {
+            final Mono<ServiceBusReceiveLink> createLink = connectionCacheWrapper.getConnection()
+                .flatMap(connection -> connection.createReceiveLink(linkName(sessionId), entityPath, receiveMode, null,
+                    entityType, identifier, 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)));
+            });
+        });
+    }
+
+    /**
+     * Publish the session acquire error using a bounded-elastic Thread.
+     * <p>
+     * The link-endpoint-state publisher ({@link ReceiveLinkHandler2#getEndpointStates()}) will emit error on the QPid
+     * Thread, which is a non-block-able Thread. Publishing the error on the (block-able) bounded-elastic Thread will free
+     * QPid Thread and to allow any blocking operation that downstream may do. If library do not publish in bounded-elastic
+     * Thread and downstream happens to make a blocking call on non-block-able QPid Thread then reactor-core will error
+     * - 'IllegalStateException(..*operation* are blocking, which is not supported in thread ...').
+     * </p>
+     *
+     * @param sessionId the session id.
+     * @param t the error to publish.
+     * @param logAtInfo indicates if session acquire error should be logged at "info" level from the current thread, most of
+     *     the time, the broker timeout is the reason for session acquisition failure, in case, the acquire fails
+     *     due to any other reasons, that least expected error is logged in the "info" level.
+     * @return a Mono that publishes the given error using a bounded-elastic Thread.
+     * @param <T> the type
+     */
+    private <T> Mono<T> publishError(String sessionId, Throwable t, boolean logAtInfo) {
+        final long id = System.nanoTime();
+        if (logAtInfo) {
+            logger.atInfo()
+                .addKeyValue(ENTITY_PATH_KEY, entityPath)
+                .addKeyValue(TRACKING_ID_KEY, id)
+                .log("Unable to acquire session '{}'.", sessionName(sessionId), t);
+        }
+        return Mono.<T>error(t)
+            .publishOn(Schedulers.boundedElastic())
+            .doOnError(ignored -> logger.atVerbose()
+                .addKeyValue(TRACKING_ID_KEY, id)
+                .log("Emitting session acquire error" + (logAtInfo ? "." : ": " + t.getMessage())));
     }
 
-    private Mono<ServiceBusReceiveLink> createSessionReceiveLink(String sessionId) {
-        final String linkName = (sessionId != null) ? sessionId : StringUtil.getRandomString("session-");
-        return connectionCacheWrapper.getConnection()
-            .flatMap(connection -> connection.createReceiveLink(linkName, entityPath, receiveMode, null, entityType,
-                identifier, sessionId));
+    /**
+     * Check if the given error is a remote link detach with '{errorCondition:com.microsoft:timeout}' indicating the broker
+     * waited for N seconds (60 sec default) but there was no free or new session.
+     *
+     * @param t the error to test.
+     * @return {@code true} if the error represents broker timeout.
+     */
+    private static boolean isBrokerTimeoutError(Throwable t) {
+        return t instanceof AmqpException
+            && ((AmqpException) t).getErrorCondition() == AmqpErrorCondition.TIMEOUT_ERROR;
+    }
+
+    /**
+     * Checks if the given error is a timeout error.
+     *
+     * @param t the error to test.
+     * @return {@code true} if the error represents timeout.
+     */
+    private static boolean isTimeoutError(Throwable t) {
+        return t instanceof TimeoutException || isBrokerTimeoutError(t);
+    }
+
+    /**
+     * Obtain the name for the AMQP link that channels messages from a session.
+     *
+     * @param sessionId the session to channel messages from.
+     * @return name for the AMQP link.
+     */
+    private static String linkName(String sessionId) {
+        return (sessionId != null) ? sessionId : StringUtil.getRandomString("session-");
+    }
+
+    /**
+     * Get the session name for simple local logging purpose.
+     *
+     * @param sessionId the unique id of the session or {@code null}, if session id is unknown.
+     * @return the session name.
+     */
+    private static String sessionName(String sessionId) {
+        return sessionId == null ? "unnamed" : sessionId;
     }
 
     /**