Azure Service Bus .NET SDK: Queues, Topics/Subscriptions, Sessions and ServiceBusProcessor

Version note: The code in this post was built with .NET 10, Azure.Messaging.ServiceBus 7.21.0 and Microsoft.Extensions.Azure 1.14.1, and run against the Azure Service Bus emulator.

Guess First

The orders queue has a lock duration of 30 seconds. A ServiceBusProcessor receives a message, the handler runs for 40 seconds and finally calls CompleteMessageAsync. The processor has MaxAutoLockRenewalDuration = TimeSpan.Zero. Does CompleteMessageAsync succeed? If not, where is the message, and what is its DeliveryCount on the next delivery?

Second question: the same queue has MaxDeliveryCount set to 3. The handler throws an exception for every message and neither completes nor abandons it. How many seconds does it take for the message to land in the dead-letter queue: 90 seconds (3 × the lock duration), or less?

Keeping ServiceBusClient a Singleton, and Identity

In RabbitMQ we kept the connection long-lived and opened channels per unit of work. In the Azure Service Bus SDK (Software Development Kit) the counterpart is ServiceBusClient: it manages the AMQP (Advanced Message Queuing Protocol) connection, is thread-safe, and should live as a single instance for the lifetime of the application. The ServiceBusSender, ServiceBusReceiver and ServiceBusProcessor it creates are also thread-safe and can be cached. We register it with AddAzureClients from the Microsoft.Extensions.Azure package:

using Azure.Identity;
using Bookstore.Billing.ServiceBus;
using Microsoft.Extensions.Azure;

HostApplicationBuilder builder = Host.CreateApplicationBuilder(args);

builder.Services.AddAzureClients(clients =>
{
    string? connectionString = builder.Configuration.GetConnectionString("ServiceBus");
    if (!string.IsNullOrEmpty(connectionString))
    {
        // local development: the emulator connection string
        clients.AddServiceBusClient(connectionString);
    }
    else
    {
        // Azure: no secret, managed identity or the developer's Azure CLI session
        clients.AddServiceBusClientWithNamespace(builder.Configuration["ServiceBus:Namespace"]!);
        clients.UseCredential(new DefaultAzureCredential());
    }
});

builder.Services.AddSingleton<InvoiceService>();
builder.Services.AddHostedService<OrderPlacedProcessor>();

await builder.Build().RunAsync();

AddAzureClients registers ServiceBusClient as a singleton and disposes it when the container shuts down. In Azure, use Microsoft Entra ID authentication instead of a connection string: the namespace name (like bookstore.servicebus.windows.net) is not a secret, and the application's managed identity is granted the "Azure Service Bus Data Sender" and "Azure Service Bus Data Receiver" roles. DefaultAzureCredential finds the Azure CLI (Command Line Interface) session on a development machine and the managed identity in Azure. A note: the Azure Identity documentation also recommends using a specific credential such as ManagedIdentityCredential directly in production to make the credential chain predictable; the delays while each step of the chain is tried can be annoying at startup.

If you see RootManageSharedAccessKey in a connection string, be careful: it is a key that can do anything on the namespace. That is fine for the emulator; in Azure, never give that key to an application.

The Service Bus Emulator Locally

Microsoft's official Service Bus emulator is a Linux container that keeps its data in a SQL (Structured Query Language) Server database. Every output in this post comes from this emulator. The setup is two containers:

docker network create mqwork1-net
docker run -d --name mqwork1-sql --network mqwork1-net -e ACCEPT_EULA=Y -e MSSQL_SA_PASSWORD="$SQL_PASSWORD" mcr.microsoft.com/mssql/server:2022-latest
docker run -d --name mqwork1-sbemu --network mqwork1-net -p 17220:5672 -p 17221:5300 -e SQL_SERVER=mqwork1-sql -e MSSQL_SA_PASSWORD="$SQL_PASSWORD" -e ACCEPT_EULA=Y -v "$PWD/sbemu/Config.json:/ServiceBus_Emulator/ConfigFiles/Config.json" mcr.microsoft.com/azure-messaging/servicebus-emulator:latest

Queues, topics and subscriptions are defined in Config.json. For Bookstore there is an orders queue, a session-enabled order-status queue, and a bookstore-orders topic with billing, notification and audit subscriptions. A shortened definition of the Billing subscription:

{
  "Name": "billing",
  "Properties": {
    "LockDuration": "PT30S",
    "MaxDeliveryCount": 3,
    "RequiresSession": false
  },
  "Rules": [
    {
      "Name": "order-placed-only",
      "Properties": {
        "FilterType": "Correlation",
        "CorrelationFilter": { "Label": "OrderPlaced" }
      }
    }
  ]
}

When the emulator starts it prints the connection string in its log; since we changed the port, we only need to add it to the endpoint: Endpoint=sb://localhost:17220;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=SAS_KEY_VALUE;UseDevelopmentEmulator=true;. Know the emulator's limits: entities and messages do not survive a container restart, partitioned entities are not supported, the message size is at most 256 KB, there are at most 10 concurrent connections per namespace, and only AMQP over TCP (Transmission Control Protocol) is available, no WebSockets. In Microsoft's words, the emulator is for sequential development and test scenarios; do not load test against it.

Sending, Topics and Subscription Filters

When building the message, we fill in the counterparts of the RabbitMQ properties:

static ServiceBusMessage ToMessage(OrderPlaced order) => new(BinaryData.FromObjectAsJson(order, MessageJson.Options))
{
    MessageId = Guid.NewGuid().ToString(),
    CorrelationId = order.OrderId.ToString(),
    Subject = "OrderPlaced",
    ContentType = "application/json",
    ApplicationProperties = { ["type"] = OrderPlaced.MessageType, ["total"] = (double)order.Total }
};

await using ServiceBusSender sender = client.CreateSender("bookstore-orders");
await sender.SendMessageAsync(ToMessage(NewOrder(1220.50m)));

// a batch instead of sending one by one: the SDK computes the size limit
using ServiceBusMessageBatch batch = await sender.CreateMessageBatchAsync();
foreach (decimal total in new[] { 450m, 89.90m, 2400m })
{
    if (!batch.TryAddMessage(ToMessage(NewOrder(total))))
    {
        throw new InvalidOperationException("Message too large for the batch");
    }
}
await sender.SendMessagesAsync(batch);

SendMessageAsync completes when the broker has durably accepted the message; the work we did separately for publisher confirmations in RabbitMQ is done by the SDK itself here. CreateMessageBatchAsync reads the size limit from the entity, and TryAddMessage returns false once the limit is exceeded. Real output:

Batch: 3 messages, 1446 bytes, max 262144 bytes
Sent 4 OrderPlaced + 1 OrderCancelled

We sent four OrderPlaced and one OrderCancelled to the topic. Each subscription is a virtual queue holding its own copy of the topic's messages, and its rules decide which messages it receives. Billing and Notification have a Subject = 'OrderPlaced' correlation filter, Audit has a total > 1000 SQL filter. Peeking at the subscriptions:

billing      4 messages: OrderPlaced/1220.5, OrderPlaced/450, OrderPlaced/89.9, OrderPlaced/2400
notification 4 messages: OrderPlaced/1220.5, OrderPlaced/450, OrderPlaced/89.9, OrderPlaced/2400
audit        2 messages: OrderPlaced/1220.5, OrderPlaced/2400

OrderCancelled entered no subscription, and Audit received only the two orders above 1,000 TL. Filters cannot look at the body; they only see system properties (Subject, CorrelationId…) and ApplicationProperties. That is why we also carry total as a property. Use correlation filters when you can: according to Microsoft's guidance they are evaluated more efficiently than SQL filters, and in the emulator the per-topic SQL filter quota (70) is far lower than the correlation filter quota (1,000).

Consuming with ServiceBusProcessor

Billing consumes the billing subscription with a ServiceBusProcessor inside a BackgroundService:

public sealed class OrderPlacedProcessor(
    ServiceBusClient client, InvoiceService invoices, ILogger<OrderPlacedProcessor> logger) : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        await using ServiceBusProcessor processor = client.CreateProcessor("bookstore-orders", "billing",
            new ServiceBusProcessorOptions
            {
                AutoCompleteMessages = false,
                MaxConcurrentCalls = 4,
                PrefetchCount = 8,
                MaxAutoLockRenewalDuration = TimeSpan.FromMinutes(5)
            });

        processor.ProcessMessageAsync += HandleAsync;
        processor.ProcessErrorAsync += args =>
        {
            logger.LogError(args.Exception, "Service Bus error from {Source} on {EntityPath}",
                args.ErrorSource, args.EntityPath);
            return Task.CompletedTask;
        };

        await processor.StartProcessingAsync(stoppingToken);
        logger.LogInformation("Processing bookstore-orders/billing");

        try
        {
            await Task.Delay(Timeout.Infinite, stoppingToken);
        }
        catch (OperationCanceledException)
        {
            // normal shutdown
        }

        // stop taking new messages, wait for in-flight ones to finish
        await processor.StopProcessingAsync(CancellationToken.None);
    }

    private async Task HandleAsync(ProcessMessageEventArgs args)
    {
        ServiceBusReceivedMessage message = args.Message;
        OrderPlaced? order;
        try
        {
            order = message.Body.ToObjectFromJson<OrderPlaced>(MessageJson.Options);
        }
        catch (JsonException ex)
        {
            // move a malformed message to the DLQ right away, with the reason attached
            await args.DeadLetterMessageAsync(message, "MalformedJson", ex.Message, args.CancellationToken);
            return;
        }

        await invoices.CreateInvoiceAsync(order!, args.CancellationToken);
        await args.CompleteMessageAsync(message, args.CancellationToken);
        logger.LogInformation("Completed {MessageId} (seq {Seq}, delivery {Count})",
            message.MessageId, message.SequenceNumber, message.DeliveryCount);
    }
}

The processor handles the receive loop, concurrency, lock renewal and reconnecting on transient errors for you. The option defaults, from a real run:

ReceiveMode=PeekLock AutoCompleteMessages=True MaxConcurrentCalls=1 PrefetchCount=0 MaxAutoLockRenewalDuration=00:05:00
Session: MaxConcurrentSessions=8 MaxConcurrentCallsPerSession=1

We turn off AutoCompleteMessages and complete the message by hand at the end of the handler, so the relationship between "invoice created" and "message deleted" is visible in the code. MaxConcurrentCalls = 4 runs four handlers at once, and PrefetchCount fetches the next messages ahead of time. Be careful with prefetch: a prefetched message's lock starts ticking as soon as it arrives, and the processor does not renew locks of messages waiting in the prefetch buffer. Do not make prefetch larger than a few times MaxConcurrentCalls. The real log for four orders:

info: Bookstore.Billing.ServiceBus.OrderPlacedProcessor[0]
      Completed d8bdd607-dce4-473f-a78f-6bd3da0df5f7 (seq 9, delivery 1)
info: Bookstore.Billing.ServiceBus.OrderPlacedProcessor[0]
      Completed a3b78dac-7f2f-415f-8b14-145da9b3843b (seq 10, delivery 1)
info: Bookstore.Billing.ServiceBus.OrderPlacedProcessor[0]
      Completed 3622a890-9b28-41c3-8bd5-6fb37eb825c2 (seq 11, delivery 1)
info: Bookstore.Billing.ServiceBus.OrderPlacedProcessor[0]
      Completed 402ea0ce-5699-4705-b61f-6285d123caa5 (seq 12, delivery 1)

For a message containing malformed JSON (JavaScript Object Notation), the handler moved it to the dead-letter queue right away; reading the DLQ (Dead Letter Queue) shows the reason next to it:

DLQ: MessageId=bad-json-1 Reason=MalformedJson Description=Expected depth to be zero at the end of the JSON payload. There is an open JSON object or array that should be closed. Path: $.orderId | LineNumber: 0 | BytePositionInLine: 12.

PeekLock, Lock Duration and Lock Renewal

In PeekLock mode a received message is not deleted; it is locked for a certain time and becomes invisible to other consumers. Within that time one of four answers is given: Complete (delete), Abandon (release the lock, deliver again), DeadLetter (move to the DLQ) or Defer (set aside to be read later by sequence number). The lock duration is an entity setting; in Azure the default is 1 minute and the maximum is 5 minutes. In ReceiveAndDelete mode the message is deleted on delivery; like autoAck: true in RabbitMQ, it is lost if the process crashes.

The first question at the top. A 30-second lock, a 40-second handler, lock renewal off:

== no renew
  0.1s received DeliveryCount=1 LockedUntil=02:14:49
 40.4s ServiceBusException Reason=MessageLockLost
 40.5s redelivered DeliveryCount=2
== renew
  0.2s received DeliveryCount=1 LockedUntil=02:15:30
 40.2s completed
 45.2s queue empty

The lock expired at 30 seconds, the message went back to the queue and its DeliveryCount increased. The work done by the 40th second was wasted, CompleteMessageAsync blew up with MessageLockLost, and the message was delivered again. For Billing this means two invoices for the same order. With the default MaxAutoLockRenewalDuration (5 minutes), the processor renewed the lock in the background and the work finished cleanly. But renewal is not a guarantee: during a network outage the renewal request can fail too. If the work takes close to the lock duration, either make the work smaller or make the handler idempotent; we will build the latter in the series' idempotent consumer post.

Second question: the handler throws an exception.

  0.2s attempt DeliveryCount=1
  0.2s ProcessError source=ProcessMessageCallback: Invoice service unavailable
  0.2s attempt DeliveryCount=2
  0.2s ProcessError source=ProcessMessageCallback: Invoice service unavailable
  0.2s attempt DeliveryCount=3
  0.2s ProcessError source=ProcessMessageCallback: Invoice service unavailable
DLQ: MessageId=poison-1 Reason=MaxDeliveryCountExceeded Description=Message could not be consumed after 3 delivery attempts.

Not 90 seconds, but within the same second. When the handler throws and has not settled the message, the processor abandons it automatically (this holds even with AutoCompleteMessages = false), the message comes back immediately, and the three attempts are used up within milliseconds. For a transient error, that is not "wait and retry" but "hit the wall three times in a row and land in the DLQ". We will handle delayed retry (resending as a scheduled message, or waiting before abandoning) in the series' retry post; what to remember here is that MaxDeliveryCount alone is not a retry strategy. Its default in Azure is 10.

Ordered Processing with Sessions

An order's status changes (Paid, Packed, Shipped) must be processed in order, but different orders can be processed in parallel. In an entity with RequiresSession enabled, every message carries a SessionId; messages of the same session are delivered in order and to only one receiver at a time:

foreach ((string order, string status) in new[] {
    (a, "Paid"), (b, "Paid"), (a, "Packed"), (b, "Packed"), (a, "Shipped"), (b, "Shipped") })
{
    await sender.SendMessageAsync(new ServiceBusMessage(status) { SessionId = order });
}

await using ServiceBusSessionProcessor processor = client.CreateSessionProcessor("order-status",
    new ServiceBusSessionProcessorOptions { MaxConcurrentSessions = 2, MaxConcurrentCallsPerSession = 1 });

  0.3s session=order-A status=Paid seq=1 thread=5
  0.4s session=order-B status=Paid seq=2 thread=9
  0.5s session=order-B status=Packed seq=4 thread=9
  0.5s session=order-A status=Packed seq=3 thread=9
  0.7s session=order-B status=Shipped seq=6 thread=5
  0.8s session=order-A status=Shipped seq=5 thread=9

The two orders interleaved, but each order's own sequence was preserved. The order id is a natural choice for SessionId. If you raise MaxConcurrentCallsPerSession above 1, you break the in-session ordering guarantee with your own hands. Sessions can also hold session state (SetSessionStateAsync), which is handy for keeping the state of a small workflow next to its messages. Why ordering is an expensive guarantee is explained in the Messaging and Event-Driven Architecture series post "Message Ordering and Partitioning".

Scheduled Messages and Duplicate Detection

Service Bus can make a message visible at a future time. For jobs like "remind the customer 24 hours later if payment has not arrived", you do not need a separate scheduler:

DateTimeOffset at = DateTimeOffset.UtcNow.AddSeconds(10);
long seq = await sender.ScheduleMessageAsync(new ServiceBusMessage("payment reminder"), at);

  0.1s scheduled seq=3 for 02:14:11
 10.1s received 'payment reminder' at 02:14:11, ScheduledEnqueueTime=02:14:11

If you keep the returned sequence number, you can cancel the reminder with CancelScheduledMessageAsync(seq) when the payment arrives.

Duplicate detection, when enabled on an entity, silently drops a second message arriving with the same MessageId within a time window. On the orders queue it is enabled with a 20-second window:

Sent attempt 1 with MessageId 580d600b-43b1-4095-b0f4-24aefca8cf95
Sent attempt 2 with MessageId 580d600b-43b1-4095-b0f4-24aefca8cf95
Received 1: attempt 1

The second send did not fail, but it did not enter the queue. This feature resolves the "did I send it or not?" uncertainty on the producer side: after a timeout you can safely resend the same message with the same MessageId. It has two conditions. First, MessageId must stay the same on resend; if you generate a new Guid.NewGuid() on every attempt, the feature catches nothing. Microsoft's recommendation is to derive the id from the business context (such as the order number and the message subject). Second, the window is not infinite: the default is 10 minutes, the minimum 20 seconds, the maximum 7 days, and throughput drops as the window grows. Duplicate detection does not solve duplicates on the consumer side (redeliveries); that is the idempotent consumer's job.

Production Notes

Choose the lock duration based on the p99 duration of the work, not its average. For long work, rely on lock renewal, but write the handler to survive redelivery.

Watch the DLQ. A dead-lettered message does not come back on its own, and if nobody looks, it piles up silently. Set an alert on the DLQ message count in Azure Monitor. Inspecting and reprocessing the DLQ is the subject of the series' dead-letter post.

Know your tier. Topics, subscriptions and duplicate detection are not available in the Basic tier; they are in Standard and Premium. The message size is 256 KB in Standard, and Premium supports larger messages. Do not put a large PDF (Portable Document Format) inside a message; write the file to blob storage and carry its reference in the message.

Calculate the total concurrency of your processors. Five instances × MaxConcurrentCalls = 4 means twenty invoices at once; the database the invoices are written to must handle that.

Know the transport for network restrictions. If port 5671 is closed on a corporate network, ServiceBusClientOptions.TransportType = AmqpWebSockets connects over 443. The emulator does not support it; you can only try this setting in Azure.

Common Pitfalls

Creating a new ServiceBusClient per request. Each client opens a new AMQP connection; it is slow and eats into the connection quota. The client is a singleton, and senders and processors are cached too.

Setting AutoCompleteMessages = false and forgetting to complete. The message comes back when the lock expires, and when it reaches MaxDeliveryCount it lands in the DLQ without any error. You see "processed" in the logs, but the message is in the DLQ.

Work exceeding the lock duration. After MessageLockLost the work is done again; in a non-idempotent handler that means a double invoice.

Mistaking throwing an exception for a retry. As we saw, three attempts are used up in the same second.

Thinking a filter can look at the body. Put every piece of information a filter needs into ApplicationProperties.

Using duplicate detection with a new MessageId per attempt. The feature looks enabled, but catches nothing.

A Small Experiment

★☆☆ Add a SQL filter to the Billing subscription: only messages whose Currency property is 'TRY'. Remember that in the emulator you have to write it into Config.json and restart the container.

★★☆ Instead of throwing in the handler, catch the error, schedule a copy of the message 10 seconds later with ScheduleMessageAsync, and complete the original. Carry the attempt count in an ApplicationProperties field, and dead-letter the message by hand after the third attempt.

★★★ Build an order state machine with session state on the order-status session: if Shipped arrives before Paid, defer the message and get it back by sequence number when Paid arrives.

In RabbitMQ and Azure Service Bus a message is deleted once it is processed. In the next post we drop that assumption: Kafka in .NET with Confluent.Kafka, choosing a partition with the key, consumer groups, offset commits and rebalancing; a world where you manage not the message but the read position.