Fixes session pump stop when auto-renew lock task expires

[inech]

Fixes the bug when the session handler stops processing messages if renew lock duration is less than it takes to process a message in user callback.

The code flow in SessionReceivePump.cs which leads to a problem looks like this:

1. When a new session is accepted, the MessagePumpTaskAsync starts a renew lock task in background and a timer to cancel this task in 1 second (the value specified in MaxAutoRenewDuration).
2. 1 second has passed and OnUserCallBackTimeout is executed which cancels and disposes renewLockCancellationTokenSource.
3. After ~30 seconds MessagePumpTaskAsync is getting some exception and pops out to a finally block which should release a semaphore. But it couldn't do it because of exception ObjectDisposedException when trying to cancel already disposed (in step 2) renewLockCancellationTokenSource.

Reproduction program
It sends a few messages in different sessions and then tries to receive them. The gotcha here is that processing a message takes longer than LockDuration and AutorenewLockDuration.
Expected: session pump indefinitely tries to process messages without stopping.
Actual: after the first message the pump is stuck and don't try to process new messages.

using System;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using HangReproduction.SessionClient;
using Microsoft.Azure.ServiceBus;
using Microsoft.Azure.ServiceBus.Core;

namespace HangReproduction
{
    internal class Program
    {
        private static async Task Main()
        {
            // Set your connection string to a queue with sessions enabled and LockDuration set to 10 seconds
            const string connectionString = "...secret...";

            // Send some messages
            //const int messageCount = 50;
            //Console.WriteLine("Sending test messages...");
            //await SendSomeMessages(connectionString, messageCount);
            //Console.WriteLine($"{messageCount} messages have been sent.");

            // Start receiving messages
            StartReceiveHandler(connectionString);

            Console.WriteLine("Press any key to exit...");
            Console.ReadKey();
        }

        private static void StartReceiveHandler(string connectionString)
        {
            var sessionClient = new Microsoft.Azure.ServiceBus.QueueClient(new ServiceBusConnectionStringBuilder(connectionString));
            sessionClient.RegisterSessionHandler(MessageHandler, new SessionHandlerOptions(ExceptionReceivedHandler)
            {
                // Set only 1 concurrent session to faster reveal the bug with pumps
                MaxConcurrentSessions = 1,
                MaxAutoRenewDuration = TimeSpan.FromSeconds(1) // Should be less than the time to process a message and LockDuration
            });
        }

        private static async Task MessageHandler(IMessageSession session, Message message, CancellationToken cancellationToken)
        {
            Console.WriteLine($"{DateTime.Now} Session {session.SessionId} message processing begin...");

            // Wait longer than MaxAutoRenewDuration
            await Task.Delay(TimeSpan.FromSeconds(20), cancellationToken);

            Console.WriteLine($"{DateTime.Now} Session {session.SessionId} message processing end");
        }

        private static Task ExceptionReceivedHandler(ExceptionReceivedEventArgs arg)
        {
            Console.WriteLine($"{DateTime.Now} Exception in handler: {arg.Exception.Message}");
            return Task.CompletedTask;
        }

        private static async Task SendSomeMessages(string connectionString, int count)
        {
            var messageSender = new MessageSender(new ServiceBusConnectionStringBuilder(connectionString));

            for (var i = 0; i < count; i++)
            {
                var message = new Message(new byte[20])
                {
                    SessionId = Guid.NewGuid().ToString()
                };
                await messageSender.SendAsync(message);
            }
        }
    }
}

P.S. this problem is already fixed in MessageReceivePump.cs#L190 but in SessionReceivePump it was overlooked.

[msftclas]

All CLA requirements met.

[azuresdkci]

Can one of the admins verify this patch?

[inech]

@nemakam Hi! Could you please review this PR or suggest somebody that could do that?
Same for #6485

[nemakam]

@inech - We will take a look and respond soon. Thanks
/cc @jsquire

[nemakam]

[jsquire]

Merging by Neeraj's request.

[petrformanek]

Hi @nemakam. When do you plan to release a new version of SDK with this fix?

[nemakam]

Hoping to release in next couple of days. All the code changes are already in place.

[lfialho]

So the current work around would be increasing the MaxAutoRenewDuration ?

[inech]

So the current work around would be increasing the MaxAutoRenewDuration ?

Hi, yes. Just make sure that your Handler doesn't process the message longer than MaxAutoRenewDuration. If you set it to 5 minutes make sure that your handler completes in less than 5 min.