We are writing an Azure function that will process many messages in batches but, for a single message in the queue, we are receiving multiple ServiceBusReceivedMessage with the same message id, and body but different delivery count (1,2,3). That is causing some issues as we would like to process batches of fixed-size to optimize costs on a dependency called by this function.
In addition to that, when we try to complete some of the messages, it will sometimes give us an Azure.Messaging.ServiceBus.ServiceBusException: "The lock supplied is invalid. Either the lock expired, or the message has already been removed from the queue". We think both problems are related since each duplicate message has a different lock token. We partially mitigated this by deduping messages by message id and doing a RenewMessageLockAsync(message) on every message before completing.
Function signature looks as below with IsBatched flag as true:
shipbob-ovelb
commented
1 month ago
Description
.Net Framework: .net 8.0 Microsoft.Azure.Functions.Worker.Extensions.ServiceBus 5.22.0
We are writing an Azure function that will process many messages in batches but, for a single message in the queue, we are receiving multiple ServiceBusReceivedMessage with the same message id, and body but different delivery count (1,2,3). That is causing some issues as we would like to process batches of fixed-size to optimize costs on a dependency called by this function. In addition to that, when we try to complete some of the messages, it will sometimes give us an Azure.Messaging.ServiceBus.ServiceBusException: "The lock supplied is invalid. Either the lock expired, or the message has already been removed from the queue". We think both problems are related since each duplicate message has a different lock token. We partially mitigated this by deduping messages by message id and doing a RenewMessageLockAsync(message) on every message before completing.
Function signature looks as below with IsBatched flag as true:
[Function("TestFunction")] public async Task RunQueueAsync( [ServiceBusTrigger("ourtopic", "oursubscription", Connection = "ServiceBusConfiguration:ConnectionString", **IsBatched = true, AutoCompleteMessages = false**)] ServiceBusReceivedMessage[] messages, ServiceBusMessageActions messageActions)Host.json:
{ "version": "2.0", "logging": { "applicationInsights": { "samplingSettings": { "isEnabled": true, "excludedTypes": "Request" }, "enableLiveMetricsFilters": true } }, "functionTimeout": "00:30:00", "extensions": { "serviceBus": { "autoCompleteMessages": false, "autoRenewTimeout": "00:05:00", "maxBatchWaitTime": "00:02:00", "maxMessageBatchSize": 800, "minMessageBatchSize": 80, "maxAutoLockRenewalDuration": "00:30:00" } } }Logs:
2024-09-17T21:30:13.837 [Information] Executing 'Functions.TestFunction' (Reason='(null)', Id=0d0777fe-d8f8-4ef9-a377-88f671929505) 2024-09-17T21:30:13.838 [Information] Trigger Details: MessageIdArray: 4ab8a173bf4a414b9bebd81ba5816eb3,4ab8a173bf4a414b9bebd81ba5816eb3,4ab8a173bf4a414b9bebd81ba5816eb3, SequenceNumberArray: 6,6,6, DeliveryCountArray: 1,2,3, EnqueuedTimeUtcArray: 2024-09-17T21:28:13.7660000+00:00,2024-09-17T21:28:13.7660000+00:00,2024-09-17T21:28:13.7660000+00:00, LockedUntilArray: 2024-09-17T21:29:13.7970000+00:00,2024-09-17T21:30:13.7700000+00:00,2024-09-17T21:31:13.7110000+00:00, SessionId: (null)Steps to reproduce