[BUG] ServiceBusSessionReceiverClient closes only after timeout

[pawel-wilczopolski]

Describe the bug
After invoking ServiceBusSessionReceiverClient close() method, the operation lasts for the tryTimeout period (1 minute by default) of AmqpRetryOptions.
The client should be closed immediately instead of waiting 1 minute.

Exception or Stack Trace
The following log is always visible after timeout (default 1 minute) is reached:
"Timed out waiting for RequestResponseChannel to complete closing. Manually closing."

To Reproduce

1. Accept any session via ServiceBusSessionReceiverClient.
2. Invoke 'close()' method on both ServiceBusSessionReceiverClient and ServiceBusReceiverClient.
3. ServiceBusSessionReceiverClient will be closing 1 minute.

Code Snippet

   public Mono<Void> closeAsync(){
        final Mono<Void> closeOperationWithTimeout  = closeMono.asMono().timeout(retryOptions.getTryTimeout()).onErrorResume(TimeoutException.class, error -> {
            return Mono.fromRunnable(() -> {

                logger.info("Timed out waiting for RequestResponseChannel to complete closing. Manually closing.");

                onTerminalState("SendLinkHandler");
                onTerminalState("ReceiveLinkHandler");
            });
        }).subscribeOn(Schedulers.boundedElastic());

    if (isDisposed.getAndSet(true)) {
        logger.verbose("Channel already closed.");
        return closeOperationWithTimeout;
    }

    logger.verbose("Closing request/response channel.");

    return Mono.fromRunnable(() -> {
        try {
            // Schedule API calls on proton-j entities on the ReactorThread associated with the connection.
            provider.getReactorDispatcher().invoke(() -> {
                logger.verbose("Closing send link and receive link.");

                sendLink.close();
                receiveLink.close();
            });
        } catch (IOException | RejectedExecutionException e) {
            logger.info("Unable to schedule close work. Closing manually.");

            sendLink.close();
            receiveLink.close();
        }
    }).subscribeOn(Schedulers.boundedElastic()).then(closeOperationWithTimeout);

Expected behavior
The ServiceBusSessionReceiverClient should be closed immediately.

Setup (please complete the following information):

• OS: Windows 11
• IDE: IntelliJ IDEA
• Library/Libraries: azure-messaging-servicebus:7.15.2
• Java version: 11

[github-actions]

@anuchandy @conniey @lmolkova

[github-actions]

Thank you for your feedback. Tagging and routing to the team member best able to assist.

[anuchandy]

Hello @pawel-wilczopolski hope you're doing good. @conniey and I discussed this today. So, the internal state machine of AMQP library is asynchronous and the reason for it wait on force close is to ensure any lingering messages, dispositions (complete|abandon) enqueued and other state in transit is processed, so things are cleaned up gracefully.

If a continuous receive and high throughput is desired, it is recommended to use the ServiceBusProcessorClient. Please see here. As it says there, the use of Synchronous Client (ServiceBus[Session|Receiver]Client) is typically discouraged, because the user has to deal with more errors and possible message expirations. On the other hand, the ServiceBusProcessorClient handles errors automatically and takes care of multi-threading, so receive from multiple sessions can happen in parallel. For the long term, ServiceBusProcessorClient will help the applications be more reliable.

If Synchronous client is still needed, and if you’re going to build a new ServiceBusSessionClient instance or restart the service after the close, you may do a fire-and-forget close, i.e, invoke close on a separate thread and giving it a few seconds before the restart/building new client using a new builder. The code may looks like-

// The ServiceBusSessionReceiver is no longer needed.
// Fire and forget the client close
try {
CompletableFuture.supplyAsync(() -> {
    // call ServiceBusSessionReceiver::close.
    return null;
})
.toCompletableFuture().get(5, TimeUnit.SECONDS);
} catch (Exception e) {
    // log(e);
}
// Service restarts or application creates new ServiceBusSessionReceiver

[pawel-wilczopolski]

Hi all, I cannot use ServiceBusProcessorClient because it hides a lot of errors that I need to handle on my side. Your solution is not resolving the real issue. I found out that you have a bug because the waiting logic begins too fast. In other words – we start waiting 1 minute BEFORE we get the ACK from the server that would cause the graceful shutdown immediately. Please refer to this. Regards, Paweł Wilczopolski From: Anu Thomas Chandy ***@***.***> Sent: Thursday, April 11, 2024 12:43 AM To: Azure/azure-sdk-for-java ***@***.***> Cc: Pawel Wilczopolski ***@***.***>; Mention ***@***.***> Subject: Re: [Azure/azure-sdk-for-java] [BUG] ServiceBusSessionReceiverClient closes only after timeout (Issue #39565) Hello @pawel-wilczopolski<https://urldefense.proofpoint.com/v2/url?u=https-3A__github.com_pawel-2Dwilczopolski&d=DwMFaQ&c=euGZstcaTDllvimEN8b7jXrwqOf-v5A_CdpgnVfiiMM&r=HXFQ3pqBriDJ5Fy_brwlqr7slA4OrpjHCKqO4kvqFDM&m=bA1gV65wqjRe2lAi81ye89XYjbY4V3az-9PSi4Bc-o08O6PfKEIu3j4Tccv8m7jc&s=3og3Po7PP6dlUKSZJWABLmfj2SZOgJPEpZq2LTCDkys&e=> hope you're doing good. @conniey<https://urldefense.proofpoint.com/v2/url?u=https-3A__github.com_conniey&d=DwMFaQ&c=euGZstcaTDllvimEN8b7jXrwqOf-v5A_CdpgnVfiiMM&r=HXFQ3pqBriDJ5Fy_brwlqr7slA4OrpjHCKqO4kvqFDM&m=bA1gV65wqjRe2lAi81ye89XYjbY4V3az-9PSi4Bc-o08O6PfKEIu3j4Tccv8m7jc&s=4qL1Vp_FvUv4g4M_D6NxnUkXSURX_3ZeGJW3varTNSw&e=> and I discussed this today. So, the internal state machine of AMQP library is asynchronous and the reason for it wait on force close is to ensure any lingering messages, dispositions (complete|abandon) enqueued and other state in transit is processed, so things are cleaned up gracefully. If a continuous receive and high throughput is desired, it is recommended to use the ServiceBusProcessorClient. Please see here<https://urldefense.proofpoint.com/v2/url?u=https-3A__github.com_Azure_azure-2Dsdk-2Dfor-2Djava_tree_main_sdk_servicebus_azure-2Dmessaging-2Dservicebus-23when-2Dto-2Duse-2Dservicebusprocessorclient&d=DwMFaQ&c=euGZstcaTDllvimEN8b7jXrwqOf-v5A_CdpgnVfiiMM&r=HXFQ3pqBriDJ5Fy_brwlqr7slA4OrpjHCKqO4kvqFDM&m=bA1gV65wqjRe2lAi81ye89XYjbY4V3az-9PSi4Bc-o08O6PfKEIu3j4Tccv8m7jc&s=d7wScfcArik_u-c7hMieyD3tmCX4hMjBndAILklD-DQ&e=>. As it says there, the use of Synchronous Client (ServiceBus[Session|Receiver]Client) is typically discouraged, because the user has to deal with more errors and possible message expirations. On the other hand, the ServiceBusProcessorClient handles errors automatically and takes care of multi-threading, so receive from multiple sessions can happen in parallel. For the long term, ServiceBusProcessorClient will help the applications be more reliable. If Synchronous client is still needed, and if you’re going to build a new ServiceBusSessionClient instance or restart the service after the close, you may do a fire-and-forget close, i.e, invoke close on a separate thread and giving it a few seconds before the restart/building new client using a new builder. The code may looks like- // The ServiceBusSessionReceiver is no longer needed. // Fire and forget the client close try { CompletableFuture.supplyAsync(() -> { // call ServiceBusSessionReceiver::close. return null; }) .toCompletableFuture().get(5, TimeUnit.SECONDS); } catch (Exception e) { // log(e); } // Service restarts or application creates new ServiceBusSessionReceiver — Reply to this email directly, view it on GitHub<https://urldefense.proofpoint.com/v2/url?u=https-3A__github.com_Azure_azure-2Dsdk-2Dfor-2Djava_issues_39565-23issuecomment-2D2048553634&d=DwMFaQ&c=euGZstcaTDllvimEN8b7jXrwqOf-v5A_CdpgnVfiiMM&r=HXFQ3pqBriDJ5Fy_brwlqr7slA4OrpjHCKqO4kvqFDM&m=bA1gV65wqjRe2lAi81ye89XYjbY4V3az-9PSi4Bc-o08O6PfKEIu3j4Tccv8m7jc&s=h4bBCyIldjVmDWflTEJXxxFjZyKcrHdG9at1RDockUM&e=>, or unsubscribe<https://urldefense.proofpoint.com/v2/url?u=https-3A__github.com_notifications_unsubscribe-2Dauth_BFMFPJQ7PLUME3CZXDLNNTTY4W55VAVCNFSM6AAAAABFZJCK2OVHI2DSMVQWIX3LMV43OSLTON2WKQ3PNVWWK3TUHMZDANBYGU2TGNRTGQ&d=DwMFaQ&c=euGZstcaTDllvimEN8b7jXrwqOf-v5A_CdpgnVfiiMM&r=HXFQ3pqBriDJ5Fy_brwlqr7slA4OrpjHCKqO4kvqFDM&m=bA1gV65wqjRe2lAi81ye89XYjbY4V3az-9PSi4Bc-o08O6PfKEIu3j4Tccv8m7jc&s=g-c5lzdaj88g7aYYtzSlBETN2B1EqnOdSka9EEfu0CE&e=>. You are receiving this because you were mentioned.Message ID: ***@***.******@***.***>>

[anuchandy]

Adding additional notes on the safety of the above workaround provided - when offloading the closure with a client-side timeout, there are two execution paths, both leading to eventual shutdown and garbage collection of reactor-executor thread, i.e., the reactor-executor thread won’t stay around and will not have any impact.

• [Execution-Path-1] Either application or host (VM, container) exit after calling close().
• [Execution-Path-2] Or, the application calls close() and continues with creating a new Client.

In the first case, since the JVM exits, all threads get cleaned up as part of the exit. In the second case, the old reactor-executor thread will continue its final cleanup activities on the background (without blocking the main thread) and exit.

We’ve created a work item to improve this experience. Linking it here, please track it there. Closing this.