Azure / Azure/azure-functions-dotnet-worker
Service Bus Trigger does not process sessions concurrently when using batches
- 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