Investigate Force-Closing AMQP Links for Cancellation

[jsquire]

Summary

The Event Hubs and Service Bus client libraries rely on the Azure AMQP library for communication with the messaging services. The AMQP library was built around the concept of timeouts and does not offer support for cancellation tokens. This leads to scenarios where the Event Hubs or Service Bus client libraries are awaiting an AMQP operation and are unable to respond to cancellation.

One of the scenarios where this causes pain for developers is when one of the messaging processors is in use. When a caller invokes the StopProcessingAsync or StopProcessing method, the processor will attempt to gracefully shut down, stopping tasks that are reading from the service and ensuring proper clean-up. Because the AMQP communication is timeout-based, when there are no messages flowing through the system, receive operations may block while being awaited for up to the configured TryTimeout, which by default is 60 seconds. This leads to a poor user experience where the processors take a long time to shut down and may cause issues with some host environments.

In a recent conversation with the AMQP library team, it was suggested that force-closing an active AMQP link could be safely used to cancel an operation in-flight. This would potentially allow the client types to respect cancellation while also gracefully shutting down and cleaning up. Doing so would introduce additional overhead for each service operation that would only provide a benefit in cancellation cases, but warrants investigation to understand the impact.

Scope of Work

• Determine an approach to observing cancellation while awaiting a service operation. It may be possible to use the Azure.Core Task Extensions, or may take a similar form to the approach used during test scenarios:

public async Task<T> Execute<T>(Task<T> serviceTask, CancellationToken mainCancellation)
{
    using var linkedCancellation = CancellationTokenSource.CreateLinkedTokenSource(mainCancellation);   
    
    var delayTask = Task.Delay(Timeout.Infinite, linkedCancellation.Token);    
    
    var completedTask = await Task.WhenAny(
        delayTask,
        serviceTask
    ).ConfigureAwait(false);

    if (completedTask != delayTask)
    {
        linkedCancellation.Cancel();

        try
        {
            await delayTask.ConfigureAwait(false);
        }
        catch (OperationCanceledException)
        {
            // Expected
        }
        
        return (await serviceTask.ConfigureAwait(false));
    }
    
    await delayTask.ConfigureAwait(false);
    throw new TaskCanceledException("This should never execute");
}

• Investigate the performance difference of applying an abstraction, including any additional allocations.

• Assess the risk and potential for errors when refactoring.

• Determine whether it makes sense to include a central executor for retry functionality in Event Hubs..

Success Criteria

• Investigate the potential for use an approach to detecting cancellation and force-closing an AMQP link when cancelled.

• Perform a small, tightly-scoped prototype and bench mark the difference to understand characteristics.

• If the change is recommended, discuss with the .NET architects and the other .NET messaging developers to get buy-in.

• If consensus is reached on the approach and need, the issues needed to track the implementation work have been created.

[danielmarbach]

I did a quick spike. What do you think about this approach?

#19888

[danielmarbach]

Another one #19955

[jsquire]

Reopening; we still don't yet have a final call on an approach.