How to remove deferred azure service bus messages?

Question

I'm using deferred messages and manage to retrieve them and process them the way I want. Now I need a way to delete them (normal messages can be "Completed") so they don't stay forever but I can't find out how.

Here's how I retrieve the message:

var message = await ServiceBusReceiver.ReceiveDeferredMessageAsync(
    deferredMessage.SequenceNumber,
    cancellationToken
);

And this is what I tried first to delete them

await ServiceBusReceiver.CompleteMessageAsync(message, cancellationToken);

Which failed with an error claiming the lock was not valid so I tried

await ServiceBusReceiver.RenewMessageLockAsync(message, cancellationToken);
await ServiceBusReceiver.CompleteMessageAsync(message, cancellationToken);

But the error persist.

EDIT:

I created a demo:

using Azure.Identity;
using Azure.Messaging.ServiceBus;

const string ToDefer = nameof(ToDefer);
const string UnDefer = nameof(UnDefer);

const string TopicName = "demo-deferred-sj";
const string ServiceBusNamespace = "#######";
const string SubscriptionName = "all";

var oneSecondMoreThanLockDuration = TimeSpan.FromSeconds(6);

var userName = System.Environment.GetEnvironmentVariable("USERNAME");

var serviceBusClient = new ServiceBusClient(
    $"{ServiceBusNamespace}.servicebus.windows.net",
    new VisualStudioCredential()
);

var sender = serviceBusClient.CreateSender(TopicName);
var processor = serviceBusClient.CreateProcessor(TopicName, SubscriptionName, new ServiceBusProcessorOptions
{
    AutoCompleteMessages = false
});
var receiver = serviceBusClient.CreateReceiver(TopicName, SubscriptionName);

async Task SendMessagesAsync(string kind, string param = "", DateTimeOffset? scheduleTime = null)
{
    var messageId = $"{kind}|{userName}|{param}|{Guid.NewGuid()}";
    Console.WriteLine($"Sending {messageId}"
        + (scheduleTime.HasValue ? $" scheduled for {scheduleTime}" : "")
    );
    var serviceBusMessage = new ServiceBusMessage()
    {
        MessageId = messageId
    };
    if (scheduleTime.HasValue)
    {
        serviceBusMessage.ScheduledEnqueueTime = scheduleTime.Value;
    }
    await sender.SendMessageAsync(serviceBusMessage);
}

async Task ProcessMessageAsync(ProcessMessageEventArgs args)
{
    Console.WriteLine($"Handling {args.Message.MessageId} ");

    var messageParts = args.Message.MessageId.Split('|');
    var kind = messageParts[0];
    var user = messageParts[1];
    var param = messageParts[2];

    if (user != userName)
    {
        Console.WriteLine($"Caution handling message of another user: {user}");
    }

    switch (kind)
    {
        case ToDefer:
            await args.DeferMessageAsync(args.Message);
            var scheduleTime = args.Message.EnqueuedTime + oneSecondMoreThanLockDuration;
            await SendMessagesAsync(UnDefer, args.Message.SequenceNumber.ToString(), scheduleTime);
            break;
        case UnDefer:
            var deferredMessage = await receiver.ReceiveDeferredMessageAsync(long.Parse(param));
            Console.WriteLine($"Deferd message {deferredMessage.MessageId} processed");
            await args.CompleteMessageAsync(args.Message);
            try
            {
                // THIS WOULD DELETE THE MESSAGE IF THE LOCK WAS STILL ON
                await receiver.CompleteMessageAsync(deferredMessage);
            }
            catch (Exception ex)
            {
                Console.WriteLine($"Ignoring {ex.Message}");
            }

            break;
    }
}

processor.ProcessMessageAsync += ProcessMessageAsync;
processor.ProcessErrorAsync += eventArgs =>
{
    Console.WriteLine(eventArgs.Exception.Message);

    return Task.CompletedTask;
};

var cancellationTokenSource = new CancellationTokenSource();
await processor.StartProcessingAsync(cancellationTokenSource.Token);

await SendMessagesAsync(ToDefer);

Console.ReadKey();
cancellationTokenSource.Cancel();

user_17623897 · Answer

According to this SO discussion, Deferred messages can also be returned by peeking messages from a queue or subscription. You may acquire the sequence numbers of the deferred messages this way, and then process or delete them.
Please refer the above given link for more information.

user_236254 · Answer

I was able to execute your demo code and process all messages as expected. I was also able to create a ServiceBusException by ensuring the CompleteMessageAsync was called after the 5 second lock duration.

The ServiceBusException message was:

Ignoring The lock supplied is invalid. Either the lock expired, or the message has already been removed from the queue, or was received by a different receiver instance. (MessageLockLost)

Is it possible that your subscription's lock duration is less than the time it takes to process the message? I would try increasing the lock duration to ensure the processing completes before the lock expires.

You can execute a method like this to process any lingering deferred messages, the ones you no longer have SequenceIds for.

async Task ProcessDeferred(int maximumMessageToSearch)
{
    var messages = await receiver.PeekMessagesAsync(maximumMessageToSearch);

    if (messages != null && messages.Any())
    {
        foreach (var deferred in messages.Where(m => m.State == ServiceBusMessageState.Deferred))
        {
            Console.WriteLine($"Processing deferred message {deferred.MessageId}");
            var message = await receiver.ReceiveDeferredMessageAsync(deferred.SequenceNumber);

            // { add you processing here }

            await receiver.CompleteMessageAsync(message);

            Console.WriteLine($"Deferred message {deferred.MessageId} has been processed.");
        }
    }
}

Not being sure how closely your demo code matches your actual code, if you are using a separate time delayed message to cause trigger the processing of the deferred message, I think there is a better pattern to accomplish your goals but I'll address that thought after this - modify the handling of the Undefer portion in your switch statement to update the trigger message AFTER you know if the deferred message was processed successfully. Something like this:

switch (kind)
{
    case ToDefer:
        await args.DeferMessageAsync(args.Message);
        var scheduleTime = args.Message.EnqueuedTime + oneSecondMoreThanLockDuration;
        await SendMessagesAsync(UnDefer, args.Message.SequenceNumber.ToString(), scheduleTime);
        break;
    case UnDefer:
        var deferredMessage = await receiver.ReceiveDeferredMessageAsync(long.Parse(param));
        Console.WriteLine($"Deferd message {deferredMessage.MessageId} processed");
        try
        {
            // THIS WOULD DELETE THE MESSAGE IF THE LOCK WAS STILL ON
            await receiver.CompleteMessageAsync(deferredMessage);
            await args.CompleteMessageAsync(args.Message);
        }
        catch (Exception ex)
        {
            await args.AbandonMessageAsync(args.Message);
            Console.WriteLine($"Ignoring {ex.Message}");
        }

        break;
}

Now back to a couple better ways of handling deferring messages; instead of using a separate message to trigger the processing of a deferred message (as in the demo). I'll list a couple options that I think are better, first would be to send a copy of the original message that has delayed processing and marking the original as completed. This avoids the need to get deferred messages by SequenceId. The other option would be to have something similar to the ProcessDeferred method (above) that is triggered by a timer and possibly check the message timestamps to determine if it should wait to process until next time.

Hopefully this helps.

How to remove deferred azure service bus messages?

2 Answers