Service Bus Trigger does not process sessions concurrently when using batches

[kurt-mw]

Description

When using the ServiceBusTrigger with IsSessionsEnabled and IsBatched the function is processing all messages in for a single session and then waiting for the idle timeout before processing then next session. This ignores the MaxConcurrentSessions setting which we have set to the default 8.

Expected behaviour is 8 sessions processing in parallel with the next sessions being processed after SessionIdleTimeout has elapsed without any new messages.

The function is running v4 and isolated process, Microsoft.Azure.Functions.Worker.Extensions.ServiceBus version 5.22.2 (latest). Information below is from a test locally using core tools 4.0.7317 but I am seeing the same behaviour in an Azure Functions App.

[Function("BatchTest")]  
public async Task Run([ServiceBusTrigger("batch-test", "batch-test",  
        Connection = "AppConfig:ServiceBusConnectionString",  
        IsSessionsEnabled = true, IsBatched = true)]  
    ServiceBusReceivedMessage[] messages,  
    ServiceBusMessageActions messagesAction,  
    ServiceBusSessionMessageActions sessionMessageActions)  
{  
    _logger.LogInformation("START ------ Batch Processing Test {messages.Length} messages received for {SessionId}.", messages.Length, messages[0].SessionId);  
    await Task.Delay(300 * messages.Length);  
    _logger.LogInformation("FINISH ----- Batch Processing Test {messages.Length} messages received for {SessionId}.", messages.Length, messages[0].SessionId);  
}

Settings from the logs (run locally):

ServiceBusOptions
{
  "ClientRetryOptions": {
    "Mode": "Exponential",
    "TryTimeout": "00:01:00",
    "Delay": "00:00:00.8000000",
    "MaxDelay": "00:01:00",
    "MaxRetries": 3
  },
  "TransportType": "AmqpWebSockets",
  "WebProxy": "",
  "AutoCompleteMessages": true,
  "PrefetchCount": 50,
  "MaxAutoLockRenewalDuration": "00:05:00",
  "MaxConcurrentCalls": 16,
  "MaxConcurrentSessions": 8,
  "MaxConcurrentCallsPerSession": 1,
  "MaxMessageBatchSize": 1000,
  "MinMessageBatchSize": 1,
  "MaxBatchWaitTime": "00:00:30",
  "SessionIdleTimeout": "00:00:10",
  "EnableCrossEntityTransactions": false
}
SingletonOptions
{
  "LockPeriod": "00:00:15",
  "ListenerLockPeriod": "00:00:15",
  "LockAcquisitionTimeout": "10675199.02:48:05.4775807",
  "LockAcquisitionPollingInterval": "00:00:05",
  "ListenerLockRecoveryPollingInterval": "00:01:00"
}

Steps to reproduce

1. Create function app with a service bus trigger for a session enabled topic.
2. Configure trigger with IsSessionsEnabled = true, IsBatched = true
3. Send batches of messages to the topic with each batch a different session id
4. Run the function app and observe only 1 session is processed
5. Wait duration of session idle timeout
6. Observe the next session is processed

[zmerdev]

We are also facing this issue.

[dean-dot-git]

Need some traction on this. Big problem for our processing.

[czb182]

I hit this today, was very confused until I saw this. Please fix. For me i am seeing two processed a time.

[czb182]

I updated to latest sdk and still occuring

[czb182]

Is concurrency supported at all when using batching?

not sure if this is right code, but it looks like this limitation.

https://github.com/Azure/azure-sdk-for-net/blob/main/sdk/servicebus/Microsoft.Azure.WebJobs.Extensions.ServiceBus/src/Listeners/ServiceBusListener.cs#L438

[PetrPospisil]

Hi,
we workaround this concurrency issue by duplicating processing functions like this:

public class LogProcessingFunction
{
    private async Task Run(string[] messages, FunctionContext context, CancellationToken cancellationToken)
    {
        // your processing code here
    }

    [Function("LogProcessingFunction00")]
    public Task Run00(
        [ServiceBusTrigger($"%{nameof(AppOptions.LogsServiceBusQueue)}:{nameof(LogsServiceBusQueueOptions.QueueName)}%",
            Connection = $"{nameof(AppOptions.LogsServiceBusQueue)}:{nameof(LogsServiceBusQueueOptions.QueueConnectionStringListener)}",
            IsSessionsEnabled = true,
            IsBatched = true)] string[] messages,
        FunctionContext context, CancellationToken cancellationToken)
    {
        return Run(messages, context, cancellationToken);
    }

    [Function("LogProcessingFunction01")]
    public Task Run01(
        [ServiceBusTrigger($"%{nameof(AppOptions.LogsServiceBusQueue)}:{nameof(LogsServiceBusQueueOptions.QueueName)}%",
            Connection = $"{nameof(AppOptions.LogsServiceBusQueue)}:{nameof(LogsServiceBusQueueOptions.QueueConnectionStringListener)}",
            IsSessionsEnabled = true,
            IsBatched = true)] string[] messages,
        FunctionContext context, CancellationToken cancellationToken)
    {
        return Run(messages, context, cancellationToken);
    }

    [Function("LogProcessingFunction02")]
    public Task Run02(
        [ServiceBusTrigger($"%{nameof(AppOptions.LogsServiceBusQueue)}:{nameof(LogsServiceBusQueueOptions.QueueName)}%",
            Connection = $"{nameof(AppOptions.LogsServiceBusQueue)}:{nameof(LogsServiceBusQueueOptions.QueueConnectionStringListener)}",
            IsSessionsEnabled = true,
            IsBatched = true)] string[] messages,
        FunctionContext context, CancellationToken cancellationToken)
    {
        return Run(messages, context, cancellationToken);
    }

    [Function("LogProcessingFunction03")]
    public Task Run03(
        [ServiceBusTrigger($"%{nameof(AppOptions.LogsServiceBusQueue)}:{nameof(LogsServiceBusQueueOptions.QueueName)}%",
            Connection = $"{nameof(AppOptions.LogsServiceBusQueue)}:{nameof(LogsServiceBusQueueOptions.QueueConnectionStringListener)}",
            IsSessionsEnabled = true,
            IsBatched = true)] string[] messages,
        FunctionContext context, CancellationToken cancellationToken)
    {
        return Run(messages, context, cancellationToken);
    }

   // XX up to desired XX-concurency 
}

So if you need 8 concurrency you run 00-07 functions like this. Did it help?

[marnilss]

I was really surprised by this limitation, and also experiences it first hand. We're running multiple instances of our Functions-application in AKS (as a Docker container) and only 1 of them are receiving messages with a batched service bus trigger, which is receiving 1 batch at a time (never in parallel).