From 417fec5e736c02828b01212cb9ef55074ebd80e6 Mon Sep 17 00:00:00 2001 From: Davide Montanari Date: Thu, 22 Oct 2026 14:27:52 -0700 Subject: [PATCH 1/3] Log messages dropped by MessageSender.SendAsync when they don't fit in the batch --- .../Abstraction/ServiceBusAbstraction.cs | 36 ++++++++++++++++--- 1 file changed, 31 insertions(+), 5 deletions(-) diff --git a/src/DurableTask.ServiceBus/Common/Abstraction/ServiceBusAbstraction.cs b/src/DurableTask.ServiceBus/Common/Abstraction/ServiceBusAbstraction.cs index 26cecadc..75826549 100644 --- a/src/DurableTask.ServiceBus/Common/Abstraction/ServiceBusAbstraction.cs +++ b/src/DurableTask.ServiceBus/Common/Abstraction/ServiceBusAbstraction.cs @@ -3,8 +3,10 @@ namespace DurableTask.ServiceBus.Common.Abstraction { using System; using System.Collections.Generic; + using System.Diagnostics; using System.Linq; using System.Threading.Tasks; + using DurableTask.Core.Tracing; /// public class IMessageSession @@ -330,15 +332,39 @@ public async Task SendAsync(Message message) await sender.SendMessageAsync(message); } + /// + /// Sends the messages in a single batch. Messages that do not fit in the batch are not sent and are traced as errors. + /// public async Task SendAsync(IEnumerable messageList) { - var batch = await sender.CreateMessageBatchAsync(); - foreach (var message in messageList) + using (Azure.Messaging.ServiceBus.ServiceBusMessageBatch batch = await sender.CreateMessageBatchAsync()) { - batch.TryAddMessage(message); + List droppedMessageIds = null; + foreach (Azure.Messaging.ServiceBus.ServiceBusMessage sbMessage in messageList) + { + if (!batch.TryAddMessage(sbMessage)) + { + if (droppedMessageIds == null) + { + droppedMessageIds = new List(); + } + + droppedMessageIds.Add(sbMessage.MessageId ?? ""); + } + } + + if (droppedMessageIds != null) + { + TraceHelper.Trace( + TraceEventType.Error, + "MessageSender-SendAsync-MessagesDropped", + () => $"{droppedMessageIds.Count} message(s) did not fit in the batch for '{sender.EntityPath}' " + + $"(maximum batch size: {batch.MaxSizeInBytes} bytes, batched messages: {batch.Count}) and were not sent. " + + $"Message ids: {string.Join(", ", droppedMessageIds)}"); + } + + await sender.SendMessagesAsync(batch); } - - await sender.SendMessagesAsync(batch); } public async Task CloseAsync() From 745c131f06be02b31a4e5fe14431a60c0202ed97 Mon Sep 17 00:00:00 2001 From: Davide Montanari Date: Thu, 22 Oct 2026 14:33:59 -0700 Subject: [PATCH 2/3] Improved logging --- .../Abstraction/ServiceBusAbstraction.cs | 23 ++++++++++++------- 1 file changed, 15 insertions(+), 8 deletions(-) diff --git a/src/DurableTask.ServiceBus/Common/Abstraction/ServiceBusAbstraction.cs b/src/DurableTask.ServiceBus/Common/Abstraction/ServiceBusAbstraction.cs index 75826549..c7204029 100644 --- a/src/DurableTask.ServiceBus/Common/Abstraction/ServiceBusAbstraction.cs +++ b/src/DurableTask.ServiceBus/Common/Abstraction/ServiceBusAbstraction.cs @@ -339,28 +339,35 @@ public async Task SendAsync(IEnumerable messageList) { using (Azure.Messaging.ServiceBus.ServiceBusMessageBatch batch = await sender.CreateMessageBatchAsync()) { + int droppedCount = 0; + long droppedBodyBytes = 0; List droppedMessageIds = null; foreach (Azure.Messaging.ServiceBus.ServiceBusMessage sbMessage in messageList) { if (!batch.TryAddMessage(sbMessage)) { - if (droppedMessageIds == null) + droppedCount++; + droppedBodyBytes += sbMessage.Body.ToMemory().Length; + if (sbMessage.MessageId != null) { - droppedMessageIds = new List(); - } + if (droppedMessageIds == null) + { + droppedMessageIds = new List(); + } - droppedMessageIds.Add(sbMessage.MessageId ?? ""); + droppedMessageIds.Add(sbMessage.MessageId); + } } } - if (droppedMessageIds != null) + if (droppedCount > 0) { TraceHelper.Trace( TraceEventType.Error, "MessageSender-SendAsync-MessagesDropped", - () => $"{droppedMessageIds.Count} message(s) did not fit in the batch for '{sender.EntityPath}' " + - $"(maximum batch size: {batch.MaxSizeInBytes} bytes, batched messages: {batch.Count}) and were not sent. " + - $"Message ids: {string.Join(", ", droppedMessageIds)}"); + () => $"{droppedCount} message(s) ({droppedBodyBytes} body bytes) did not fit in the batch for '{sender.EntityPath}' " + + $"and were not sent; {batch.Count} message(s) were sent (maximum batch size: {batch.MaxSizeInBytes} bytes)." + + (droppedMessageIds != null ? $" Dropped message ids: {string.Join(", ", droppedMessageIds)}" : string.Empty)); } await sender.SendMessagesAsync(batch); From 7c7e0532084dfe302c05c40f899dc8644f7e4d3b Mon Sep 17 00:00:00 2001 From: Davide Montanari Date: Thu, 22 Oct 2026 14:43:59 -0700 Subject: [PATCH 3/3] Removed messageIds and fixed log message --- .../Common/Abstraction/ServiceBusAbstraction.cs | 13 +------------ 1 file changed, 1 insertion(+), 12 deletions(-) diff --git a/src/DurableTask.ServiceBus/Common/Abstraction/ServiceBusAbstraction.cs b/src/DurableTask.ServiceBus/Common/Abstraction/ServiceBusAbstraction.cs index c7204029..ecd049b8 100644 --- a/src/DurableTask.ServiceBus/Common/Abstraction/ServiceBusAbstraction.cs +++ b/src/DurableTask.ServiceBus/Common/Abstraction/ServiceBusAbstraction.cs @@ -341,22 +341,12 @@ public async Task SendAsync(IEnumerable messageList) { int droppedCount = 0; long droppedBodyBytes = 0; - List droppedMessageIds = null; foreach (Azure.Messaging.ServiceBus.ServiceBusMessage sbMessage in messageList) { if (!batch.TryAddMessage(sbMessage)) { droppedCount++; droppedBodyBytes += sbMessage.Body.ToMemory().Length; - if (sbMessage.MessageId != null) - { - if (droppedMessageIds == null) - { - droppedMessageIds = new List(); - } - - droppedMessageIds.Add(sbMessage.MessageId); - } } } @@ -366,8 +356,7 @@ public async Task SendAsync(IEnumerable messageList) TraceEventType.Error, "MessageSender-SendAsync-MessagesDropped", () => $"{droppedCount} message(s) ({droppedBodyBytes} body bytes) did not fit in the batch for '{sender.EntityPath}' " + - $"and were not sent; {batch.Count} message(s) were sent (maximum batch size: {batch.MaxSizeInBytes} bytes)." + - (droppedMessageIds != null ? $" Dropped message ids: {string.Join(", ", droppedMessageIds)}" : string.Empty)); + $"and were not sent; {batch.Count} message(s) were added to the batch (maximum batch size: {batch.MaxSizeInBytes} bytes)."); } await sender.SendMessagesAsync(batch);