Azure / Azure/azure-functions-dotnet-worker

Service Bus Trigger does not process sessions concurrently when using batches

Open
#3,087 7 comments 3 reactions 0 assignees View on GitHub
extensions: service-bus potential-bug
Dominant language
C#
Stars
466
Forks
215
Avg merge
3d 10h
Merged PRs (30d)
7

Description

### 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):

```json
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

Contributor guide

No contributing guide indexed for this repository

Research direction

Start with the isolated-worker ServiceBusTrigger path handling IsSessionsEnabled and IsBatched, using the provided function and ServiceBusOptions as the reproduction. Run the listed five-step scenario and compare the logs for concurrent sessions. Done means batches from up to MaxConcurrentSessions sessions process in parallel and the next session starts after SessionIdleTimeout.

Written by the indexing model from the issue text.

Assessment

Tech stack
azure, csharp
Domain
backend, distributed-systems
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Quiet
Clarity
Mostly clear
Newbie friendliness
42/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.