is there any way to force stop the ProcessMessageAsync handler when the thread keeps running in background?

ValdoG 0

My http function app tried to make a call to Service Bus to collect messages every 5 minutes and cancel after 5 seconds to response the consumer.

[FunctionName("QueryHandler")]         
public async Task<IActionResult> Run(             
[HttpTrigger(AuthorizationLevel.Function, "get", Route = null)] HttpRequest req,             
ILogger log)         
{
		async Task MessageHandler(ProcessMessageEventArgs args)
		{
		            string body = args.Message.Body.ToString();
		            var message = JsonConvert.DeserializeObject<JObject>(body);
		            Messages.Add(message);
		
		            // complete the message. Message is deleted from the queue
		            await args.CompleteMessageAsync(args.Message);
		}
		
		// handle any errors when receiving messages
		Task ErrorHandler(ProcessErrorEventArgs args)
		{
		            Console.WriteLine(args.Exception.ToString());
		            var errorObject = new JObject();
		            errorObject["ErrorMessage"] = $"Error processing the request from the queue: {args.Exception.Message}";
		            Messages.Add(errorObject);
		            return Task.CompletedTask;
		}
		
		// handler to process any errors
		processor.ProcessErrorAsync += ErrorHandler;
		
		// cancel the token after 5 secs
		tokenSource.CancelAfter(TimeSpan.FromSeconds(5));
		
		// start process async
		await processor.StartProcessingAsync(tokenSource.Token);
		
		// end the while loop for any cancelation process (5 secs)
		while (!tokenSource.IsCancellationRequested) {}
		
		// stop or end process async
		await processor.StopProcessingAsync(tokenSource.Token);

}

Issue happened when the client/consumer/postman cancel the function app calls say multiple times and it cancel gracefully the call from client/postman.

However, in function app the thread pool still running and when there's a message in the queue it will automatically pick up the message. So, when you call the function app to get the message it say no message to pick up.

Is any one experience this scenario?

To simulate:

Call function app and cancel while processing (3x)
Send message to service bus and it will automatically be pickup by the message handler.

Workaround is to stop and start the function app to resolve it.

navba-MSFT 20,890 Reputation points Microsoft Employee

2023-10-05T04:13:54.24+00:00
@ValdoG Welcome to Microsoft Q&A Forum, Thank you for posting your query here!

I understand that you are experiencing an issue where the thread pool is still running even after the client cancels the function app call. This can cause the function app to automatically pick up messages from the queue even when there are no messages to pick up.

To force stop the ProcessMessageAsync handler when the thread keeps running in the background, you can use the CancellationToken to cancel the processing of messages. In your code, you are already using the CancellationToken to cancel the token after 5 seconds. You can use the same token to cancel the processing of messages when the client cancels the function app call.

Here's an updated version of your code that cancels the processing of messages when the client cancels the function app call:

// handle any errors when receiving messages Task ErrorHandler(ProcessErrorEventArgs args) { Console.WriteLine(args.Exception.ToString()); var errorObject = new JObject(); errorObject["ErrorMessage"] = $"Error processing the request from the queue: {args.Exception.Message}"; Messages.Add(errorObject); return Task.CompletedTask; } // create a processor that we can use to process the messages var client = new ServiceBusClient("<NAMESPACE-CONNECTION-STRING>"); var processor = client.CreateProcessor("<QUEUE-NAME>", new ServiceBusProcessorOptions()); // add handler to process messages processor.ProcessMessageAsync += MessageHandler; // add handler to process any errors processor.ProcessErrorAsync += ErrorHandler; // start processing messages await processor.StartProcessingAsync(tokenSource.Token); // wait for the cancellation token to be cancelled await tokenSource.Token.WaitHandle.WaitOneAsync(); // stop processing messages await processor.StopProcessingAsync();

In this updated code, we create a CancellationTokenSource and pass its token to the StartProcessingAsync method to start processing messages**.** We then wait for the cancellation token to be cancelled using the WaitHandle.WaitOneAsync method. This method waits for the cancellation token to be cancelled or for the specified timeout to elapse. When the client cancels the function app call, the cancellation token is cancelled and the WaitHandle.WaitOneAsync method returns. We then stop processing messages using the StopProcessingAsync method.

I hope this helps!

On a side note: You can also try using the CancellationToken.ThrowIfCancellationRequested method to throw an exception when the cancellation token is requested, which can be caught in the while loop to stop the MessageHandler method. Additionally, you can also try to handle exceptions in the ErrorHandler method and use the ServiceBusProcessorOptions.MaxConcurrentCalls property to limit the number of messages that can be processed concurrently. More Info here.

Please let me know if you have any further questions.

**

Please do not forget to "Accept the answer” and “up-vote” wherever the information provided helps you, this can be beneficial to other community members.

ValdoG 0

Hi,

Thanks for the very detailed response.

I noticed system.threading don't have asynchronous WaitOneAsync
await tokenSource.Token.WaitHandle.WaitOneAsync();

namespace System.Threading{        
//        
// Summary:        
//     Gets a System.Threading.WaitHandle that is signaled when the token is canceled.        
//        
// Returns:        
//     A System.Threading.WaitHandle that is signaled when the token is canceled.        
//        
// Exceptions:        
//   T:System.ObjectDisposedException:        
//     The associated System.Threading.CancellationTokenSource has been disposed.        
public WaitHandle WaitHandle { get }
}

If I used the CancellationToken.ThrowIfCancellationRequested do I need to place this inside the message handler?

async Task MessageHandler(ProcessMessageEventArgs args)
{
        string body = args.Message.Body.ToString();
        var message = JsonConvert.DeserializeObject<JObject>(body);
        Messages.Add(message);
		
        await args.CompleteMessageAsync(args.Message);
        cancellationToken.ThrowIfCancellationRequested();
}

navba-MSFT 20,890 Reputation points Microsoft Employee

2023-10-05T08:48:21.73+00:00
@ValdoG You are correct that the WaitHandle property of the CancellationToken does not have an asynchronous WaitOneAsync method.

To wait for the cancellation token to be cancelled asynchronously, you can use the Task.FromCanceled method to create a completed task that represents the cancelled token. Here's an example:

// wait for the cancellation token to be cancelled await Task.WhenAny(Task.Delay(Timeout.Infinite, tokenSource.Token), Task.FromCanceled(tokenSource.Token));

In this code, we use the Task.Delay method to create a task that never completes, and pass the cancellation token to it. We then use the Task.FromCanceled method to create a completed task that represents the cancelled token. We use the Task.WhenAny method to wait for either the Task.Delay task or the cancelled token task to complete. When the cancelled token task completes, we know that the cancellation token has been cancelled. More Info here.

Regarding your second question, yes, you can use the CancellationToken.ThrowIfCancellationRequested method inside the message handler to throw an exception if the cancellation token has been cancelled. This will stop the processing of messages and allow the StopProcessingAsync method to be called to stop the ServiceBusProcessor instance. Here's an example:

async Task MessageHandler(ProcessMessageEventArgs args) { string body = args.Message.Body.ToString(); var message = JsonConvert.DeserializeObject<JObject>(body); Messages.Add(message); await args.CompleteMessageAsync(args.Message); // Throw an exception if the cancellation token has been cancelled cancellationToken.ThrowIfCancellationRequested(); }

ValdoG 0

I did add the tokenSource.Token.ThrowIfCancellationRequested();

However, this time the exception become OperationCanceledException and still continuing to run the message handler and consuming the new message from Service Bus. The only option to stop the background process is to stop the function app.

async Task MessageHandler(ProcessMessageEventArgs args)
{
    try
    {
        string body = args.Message.Body.ToString();
        var message = JsonConvert.DeserializeObject<JObject>(body);
        Messages.Add(message);

        // complete the message. Message is deleted from the queue
        await args.CompleteMessageAsync(args.Message);

        // Throw an exception if the cancellation token has been cancelled
        tokenSource.Token.ThrowIfCancellationRequested();
    }
    catch (TaskCanceledException)
    {
        logger.APACLogInformation($"ExecutionLog: Message handler - TaskCanceled ");
    }
    catch (OperationCanceledException)
    {
        logger.APACLogInformation($"ExecutionLog: Message handler - OperationCanceled");
    }
}

navba-MSFT 20,890 Reputation points Microsoft Employee

2023-10-05T10:25:46+00:00
@ValdoG Thanks for getting back. The OperationCanceledException is thrown when the cancellation token is cancelled, and it is expected behavior. However, it should cause the ServiceBusProcessor instance to stop processing messages and the StopProcessingAsync method to be called.

One possible reason why the message handler is still running and consuming new messages is that the StopProcessingAsync method is not being called. To ensure that the StopProcessingAsync method is called when the cancellation token is cancelled, you can add a check for the cancellation token in the MessageHandler method and call StopProcessingAsync if the token has been cancelled. Here's an example:

async Task MessageHandler(ProcessMessageEventArgs args) { try { string body = args.Message.Body.ToString(); var message = JsonConvert.DeserializeObject<JObject>(body); Messages.Add(message); // complete the message. Message is deleted from the queue await args.CompleteMessageAsync(args.Message); // Throw an exception if the cancellation token has been cancelled tokenSource.Token.ThrowIfCancellationRequested(); } catch (TaskCanceledException) { logger.APACLogInformation($"ExecutionLog: Message handler - TaskCanceled "); } catch (OperationCanceledException) { logger.APACLogInformation($"ExecutionLog: Message handler - OperationCanceled"); // Stop processing messages await processor.StopProcessingAsync(); } }

With this code, if the cancellation token has been cancelled, the message handler will stop processing messages and call StopProcessingAsync to stop the ServiceBusProcessor instance.

I hope this helps! Let me know if you have any further questions.

ValdoG 0

Thanks @navba-MSFT for patiently guiding me. Unfortunately, the message handler background service keep running even I put the await processor.StopProcessingAsync();

Btw, below is my full code...

using System;
using System.Threading.Tasks;
using Microsoft.AspNetCore.Mvc;
using Microsoft.Azure.WebJobs;
using Microsoft.Azure.WebJobs.Extensions.Http;
using Microsoft.AspNetCore.Http;
using Microsoft.Extensions.Logging;
using Newtonsoft.Json;
using Azure.Messaging.ServiceBus;
using System.Collections.Generic;
using System.Threading;
using Newtonsoft.Json.Linq;
using System.Linq;
using System.Globalization;
using Azure.Storage.Blobs;
using System.Text;

namespace TestingPOC
{
    public class QueryHandler
    {
        static string connectionString = Environment.GetEnvironmentVariable("ServiceBusConnectionString");
        static string topicName = Environment.GetEnvironmentVariable("MyTopic");

        static ServiceBusClient client;
        static ServiceBusProcessor processor;

        static List<JObject> Messages;
        static CancellationTokenSource tokenSource;

        static ILogger logger;
        string correlationId=string.Empty;

        [FunctionName("QueryHandler")]
        public async Task<IActionResult> Run(
            [HttpTrigger(AuthorizationLevel.Function, "get", Route = null)] HttpRequest req,
            ILogger log)
        {
            var subscriptionName = req.Query["sub"];

            tokenSource = new CancellationTokenSource();
            Messages = new List<JObject>();
            try
            {
                // the client that owns the connection and can be used to create senders and receivers
                client = new ServiceBusClient(connectionString);

                // the processor that reads and processes messages from the queue
                processor = client.CreateProcessor(topicName, subscriptionName, new ServiceBusProcessorOptions());

                logger = log;
                log.APACLogInformation($"ExecutionLog: Request recieved from {subscriptionName}");

                // handler to process SB message async
                processor.ProcessMessageAsync += MessageHandler;
                // handler to process any errors
                processor.ProcessErrorAsync += ErrorHandler;

                //// cancel the token after 5 secs
                tokenSource.CancelAfter(TimeSpan.FromSeconds(5));

                log.APACLogInformation($"ExecutionLog: Reading work order queue for {subscriptionName}");

                // start process async
                await processor.StartProcessingAsync(tokenSource.Token);

                //// end the while loop for any cancelation process (5 secs)
                while (!tokenSource.IsCancellationRequested) {}


                log.APACLogInformation($"ExecutionLog: Completed reading work order queue for {subscriptionName}");

                // stop or end process async
                await processor.StopProcessingAsync(tokenSource.Token);
            }
            catch (TaskCanceledException)
            {
                log.APACLogInformation($"ExecutionLog: Message reading timedout - TaskCanceled {subscriptionName}");
            }
            catch (OperationCanceledException)
            {
                log.APACLogInformation($"ExecutionLog: Message reading timedout - OperationCanceled {subscriptionName}");
            }
            catch (Exception e)
            {
                log.APACLogError(e, $"Message reading timedout {subscriptionName}");
                return new BadRequestObjectResult(e.Message);
            }
            finally
            {
                // Calling DisposeAsync on client types is required to ensure that network
                // resources and other unmanaged objects are properly cleaned up.
                if (processor != null)
                    await processor.DisposeAsync();
                if (client != null)
                    await client.DisposeAsync();
            }

            return new OkObjectResult(Messages);
        }

        // handle in receiving SB messages 
        async Task MessageHandler(ProcessMessageEventArgs args)
        {
            try
            {
                correlationId = args.Message.CorrelationId.ToString();
                logger.APACLogInformation($"ExecutionLog: MessageHandler processing with correlation ID: {correlationId}", correlationId);
                string body = args.Message.Body.ToString();
                var message = JsonConvert.DeserializeObject<JObject>(body);
                Messages.Add(message);

                // complete the message. Message is deleted from the queue
                await args.CompleteMessageAsync(args.Message);

                // Throw an exception if the cancellation token has been cancelled
                tokenSource.Token.ThrowIfCancellationRequested();
            }
            catch (TaskCanceledException)
            {
                logger.APACLogInformation($"ExecutionLog: Message handler - TaskCanceled ", correlationId);

                // Stop processing messages
                await processor.StopProcessingAsync(tokenSource.Token);
            }
            catch (OperationCanceledException)
            {
                logger.APACLogInformation($"ExecutionLog: Message handler - OperationCanceled", correlationId);

                // Stop processing messages
                await processor.StopProcessingAsync();
            }
        }

        // handle any errors when receiving messages
        Task ErrorHandler(ProcessErrorEventArgs args)
        {
            Console.WriteLine(args.Exception.ToString());
            var errorObject = new JObject();
            errorObject["ErrorMessage"] = $"Error processing the request from the queue: {args.Exception.Message}";
            Messages.Add(errorObject);
            logger.APACLogError(args.Exception, $"ExecutionLog: Error processing messages from queue {args.Exception.Message}",correlationId);
            return Task.CompletedTask;
        }

    }
}

Apparently, when the client cancellation happened the background process will only occur in the message handler it doesn't go out in that method. I think that's the reason why the processor cannot stop. However, it's really weird because supposedly it will stop once we call the stop method.

Thanks again for your support. :)

navba-MSFT 20,890 Reputation points Microsoft Employee

2023-10-06T06:13:52.4533333+00:00
@ValdoG Thanks for getting back. Please note that the CancellationToken provides a means to cancel general asynchronous processing in .NET.

The Azure Functions has built-in support for CancellationToken bindings.

Task cancellation | Microsoft Docs

Cancellation tokens - Azure Functions C# developer reference | Microsoft Docs

CancellationToken, which can be obtained by binding, is embedded in the application lifecycle and can be used to easily detect shutdowns.

public class Function1 { [FunctionName("Function1")] public async Task Run([TimerTrigger("0 */5 * * * *")] TimerInfo timer, ILogger log, CancellationToken cancellationToken) { try { log.LogInformation("Function executing"); // Simulate time-consuming processes await Task.Delay(TimeSpan.FromSeconds(30), cancellationToken); log.LogInformation("Function executed"); } catch (OperationCanceledException) { log.LogWarning("Function cancelled"); } } }

Processes that take a long time are simulated with Task.Delay. Note that Task.Delay is passed a CancellationToken.

If you send Ctrl+C to send a shutdown signal while a Function is running, the time-consuming process is canceled and a OperationCanceledException exception is caught, and the log is output.

The same process as when Ctrl+C is sent will be performed when Azure Functions are restarted or deployed, so by handling CancellationToken correctly, you can implement the termination process without breaking the integrity of your application.

Could you please update the code as above and test ? More Info here

Also did you explore the options to use the Durable Functions with terminating instances to check if that serves your requirement ?
ValdoG 0 Reputation points

2023-10-08T10:51:04.51+00:00
Thanks navba-MSFT for your suggestion. However, my function app requirements need to be call as http with get method from the consumer which need to check the service bus message and collect the payload and response back to consumer.

[FunctionName("QueryHandler")] public async Task<IActionResult> Run( [HttpTrigger(AuthorizationLevel.Function, "get", Route = null)] HttpRequest req, ILogger log, CancellationToken cancellationToken)

Now, my issue and did replicate using curl command below with Run and CTRL+C combination.

curl --location 'http://localhost:7071/api/QueryHandler' \ --header 'Content-Type: application/json'

I can see from the logger that completed the process and closed the processor and client.

But when I tried to send a message to service bus the message handler auto pick up the message and won't go to the closing of the processor and client.. From my observation it always run in the message handler method... which seems that background process didn't killed.

// handle in receiving SB messages async Task MessageHandler(ProcessMessageEventArgs args) { string body = args.Message.Body.ToString(); var message = JsonConvert.DeserializeObject<JObject>(body); Messages.Add(message); await args.CompleteMessageAsync(args.Message); }

Are you able to try this scenario?

Thanks again.
ValdoG 0 Reputation points

2023-10-09T10:16:54.2266667+00:00
Thanks again @navba-MSFT Perhaps something in my code that causes not properly disposing the cancellation..
And I just found something when I used the static for client and processor. It won't properly dispose the object and same error I encountered above.

static ServiceBusClient client; static ServiceBusProcessor processor; static CancellationTokenSource tokenSource;

However, if I remove the static, it's properly disposed the object of client and processor. The reason I used static is for async calls for scalability.

My goal is to accommodate different caller, e.g caller 1 wants to make 10 calls in 5 mins , assume token expires in 5 mins then caller 2 wants to call 5 in 5 mins.. I want to reuse the same token to the same users/callers.

just like you can use postman to generate a token , then you can use in other post or get.
navba-MSFT 20,890 Reputation points Microsoft Employee

2023-10-09T10:22:53.4+00:00

@ValdoG Thanks for your reply. Please ensure that you update all your packages (libraries) used in your project and use updated version of the func core runtime and see if that helps.
ValdoG 0 Reputation points

2023-10-09T23:38:37.71+00:00

All good. This is one of the tech debt when I joined the company.

Anyway, let me try that to uplift with latest libraries and FA core. Thank you so much for your help.

1 answer

navba-MSFT 20,890 Reputation points Microsoft Employee

2023-10-09T09:58:05.66+00:00

@ValdoG Thanks for your reply. I tried to leverage your above code and sent the GET Request from Postman and immediately cancelled it. I am able to cancel the operation just fine. See the below screenshot for your reference:

Also note that my messages are not picked from SB. Instead it is moving to Deadletter queue as expected.

Sharing the below details:

Azure Functions Core Tools

Core Tools Version: 4.0.5390 Commit hash: N/A (64-bit)

Function Runtime Version: 4.25.3.21264

Most recent version of Microsoft.Azure.WebJobs and Azure.Messaging.ServiceBus packages.
Please sign in to rate this answer.

0 comments No comments
Sign in to comment

Share via

is there any way to force stop the ProcessMessageAsync handler when the thread keeps running in background?

1 answer