Complete messages where they came from with Azure Service Bus

In the last post, we made a simplistic attempt to speed up batch completion of messages by having multiple dedicated background completion tasks. Unfortunately, this introduced an interesting side effect.

static async Task BatchCompletionLoop() {
   while(!token.IsCancellationRequested) {
         ...
         await receiveClient.CompleteBatchAsync(lockTokens).ConfigureAwait(false);
         ...
      }
      await Task.Delay(TimeSpan.FromSeconds(5), token).ConfigureAwait(false);
   }
}

The code above is using a single receive client to complete message lock tokens in batches. Those lock token could come from multiple receive clients, but they would end up being completed on the same instance. Depending on the transport type you use it might work or it won’t. When you are using the NetMessaging (also known as SBMP) transport type the above code will just work. If you switch to AMQP as the transport type, the code will blow up with the following exception

Microsoft.ServiceBus.Messaging.MessageLockLostException: The lock supplied is invalid. Either the lock expired, or the message has already been removed from the queue.
   at Microsoft.ServiceBus.Common.AsyncResult.End[TAsyncResult](IAsyncResult result)
   at Microsoft.ServiceBus.Messaging.MessageReceiver.EndCompleteBatch(IAsyncResult result)

For Azure Service Bus v3 and higher AMQP is the new standard protocol and SMBP is an opt-in choice.

The proprietary SBMP protocol that is also supported is being phased out in favor of AMQP. see documentation

What does that mean for our message completion? To have a robust completion code which works for both transport types, we have to make sure we only complete lock tokens on receivers the messages were coming from.

In its simplest form, we would need the following moving pieces. A concurrent stack or queue per message receivers, a generic message handling body which closes over the concurrent stack as an input parameter and similarly a parameterized completion loop.

var lockTokensToComplete = new ConcurrentStack<Guid>[numberOfReceivers];
// initialize the concurrent stacks
  
receiveClient1.OnMessageAsync(message => ReceiveMessage(message, lockTokensToComplete[0]);
...
receiveClientN.OnMessageAsync(message => ReceiveMessage(message, lockTokensToComplete[N-1]);

static async Task ReceiveMessage(BrokeredMessage message, ConcurrentStack<Guid> lockTokensToComplete) {
   await DoSomethingWithTheMessageAsync().ConfigureAwait(false);
   lockTokensToComplete.Push(message.LockToken);
}

and the completion loop

for(int i = 0; i < numberOfReceivers; i++) { 
   completionTasks[i] = Task.Run(() => BatchCompletionLoop(receivers[i], lockTokensToComplete[i]));
}
 
static async Task BatchCompletionLoop(MessageReceiver receiver, ConcurrentStack<Guid> lockTokensToComplete) {
   while(!token.IsCancellationRequested) {
    var lockTokens = new Guid[100];
      int numberOfItems = lockTokensToComplete.TryPopRange(lockTokens)
      if(numberOfItems > 0) {
         await receiver.CompleteBatchAsync(lockTokens).ConfigureAwait(false);
      }
      await Task.Delay(TimeSpan.FromSeconds(5), token).ConfigureAwait(false);
   }
}

We have reduced the contention since we have a dedicated lock token concurrent collection to operate on per receiver. We have a dedicated completion loop which is self-contained and only operates on the correct receiver and lock token collection pair. So far so good. Despite that we are still wasting a lot of unnecessary resources. We acquired up to the number receivers dedicated completion background tasks which might idle. And we still don’t exactly know if this code constructs scales elastically up and down depending on the number of messages received and the number of clients.

Ideally, we would have a reactive approach which automatically adapts based on parameters we give it. Building such a reactive approach will be the focus of the next few posts. Stay reactive!