CODEWITHSUNDEEP
azure.net-coreservicebusazure-servicebus-queues
Ive
Ive

Reputation: 477

Azure Servicebus queue delay in receiving ordered messages

Using an Azure ServiceBus Queue with sessions (message-ordering) enabled, I have sessions that need to last between a couple of minutes and a couple of hours.

To that end, I have configured my QueueClient as follows:

_options = new SessionHandlerOptions(ExceptionReceivedHandler)
{
    AutoComplete = false,
    MaxConcurrentSessions = 50,
    MessageWaitTimeout = TimeSpan.FromSeconds(30),
};

and start receving messages as follows:

_queueClient.RegisterSessionHandler(ProcessSessionMessagesAsync, _options);

After a few (between 1 and 6) successful (and near instantaneous) message-receive callbacks -- both for new sessions and existing ones, the receive handler just stops firing. Using ServiceBusExplorer, I can see messages sitting on the servicebus queue. Interestingly, they all have a DeliveryCount=1. After some time (this varies between a few seconds and a few minutes -- but not a multiple of MessageWaitTimeout), I start receiving a trickle of messages again. If I restart the receiver, then I sometimes get a burst of all remaining messages, sometimes nothing more.

I've tried various values for MessageWaitTimeout, and although lower values seem to lessen the issue, the delays still exist.

Interesting, if I complete sessions after every message received, the problem still persists.

Has anyone experienced anything like this? The behavior is so infuriatingly inconsistent...

FWIW, my messageReceivedHander looks something like this:

async Task ProcessSessionMessagesAsync(IMessageSession session, Message message, CancellationToken token)
{
    try
    {
        var myEvent = Serializer.Deserialize(message.Body);
        await _receiveCallback(Subscription, myEvent);
        await session.CompleteAsync(message.SystemProperties.LockToken);

        // Drop the session after every message (**makes no difference**)
        await session.CloseAsync();
    }
    catch (Exception ex)
    {
        await session.AbandonAsync(message.SystemProperties.LockToken);
    }
}

Upvotes: 1

Views: 1341

Answers (2)

Ive
Ive

Reputation: 477

This was in fact me being an idiot.

The _receiveCallback() was actually a call out to an external service (requestbin), which was throttling my messages under load and causing the behaviour I was experiencing. When I stubbed that out, I saw consistent (and fast) behvaiour.

Upvotes: 1

Jack Jia
Jack Jia

Reputation: 5559

Just some suggestions:

  1. Do not close asession in ProcessSessionMessagesAsync.

    await session.CloseAsync();

  2. As you just registered a handler, so please make your application alive to make sure that all the tasks will be finished. For example: you can end your application after a Console.Readline().

I run the official sample: BasicSessionSendReceiveUsingQueueClient, it works fine. You can have a try.

Upvotes: 0

Related Questions