Session lock expired on reading deferred message from Azure Service Bus

Viewed 30

I have 2 applications where application A is sending data sets of complex objects to application B. It is imperative that the objects are processed in a specific order foreach data set, which is why I am using a topic with sessions enabled on the Azure Service Bus, where each message has a session ID (representing a data set) and a sequence ID representing the order.

I am using a ServiceBusSessionProcessor to receive messages from the bus, it is initialized like this:

SessionOptions = new ServiceBusSessionProcessorOptions()
{
   MaxAutoLockRenewalDuration = TimeSpan.FromMinutes(5),
   AutoCompleteMessages = false,
   MaxConcurrentSessions = 4,
   ReceiveMode = ServiceBusReceiveMode.PeekLock,
   SessionIdleTimeout = TimeSpan.FromMinutes(10),
   MaxConcurrentCallsPerSession = 1,
};
SubscriptionSessionProcessor = SubscriptionClient.CreateSessionProcessor("TestTopic", "TestSubscription", SessionOptions);

When the message is received I check if the message is next in the sequence to be processed in if so I process the message otherwise I defer the message back on the bus

protected async Task HandleSessionMessageAsync<R>(SessionInformationBase sessionInformation, ServiceBusReceivedMessage message)
{
   var stateData = await sessionInformation.Message.GetSessionStateAsync();
   var sessionState = SessionStateManager.CreateSessionStateManager(stateData);

   if (sessionState.IsNextMessage(sessionInformation.SessionData.Sequence))
   {
      try
      {
         sessionInformation.ProcessorCall.Invoke();
      }
      catch (Exception e)
      {
         ExceptionTracer.TraceException(e);
      } 
      await sessionInformation.Message.CompleteMessageAsync(message);
      await HandleSessionStateAsync<R>(sessionInformation, sessionState); 
   }
   else
   {
      sessionState.DeferredMessages.Add(sessionInformation.SessionData.Sequence, message.SequenceNumber);
      //defer message directly on the message object (the message that was received on the ServiceBusSessionProcessor object
      await sessionInformation.Message.DeferMessageAsync(message);  
      await SetSessionStateAsync(sessionInformation.Message, sessionState.Serialize());
   }
}

When I process a message in sequence I check if the next message in line was already received and if so i try to read it from the bus like this:

if (!sessionState.NextMessageIsDeferred())
   return;
try
{
   var deferredMessage = await Receiver.ReceiveDeferredMessageAsync(sessionState.GetNextMessageSequenceId());
   ... //process deferred message
}

However when I call the ReceiveDeferredMessageAsync method I receive a Session lock expired exception. The session lock timeout is set to 5 minutes and I know that there were mere seconds between the deferral and reading the deferred message.

One potential issue is that I am using an object of the class ServiceBusReceiver to retrieve the deferred message, that is neither the original processor that received the message nor the object itself - but as far as I can tell those object are unable to retrieve deferred messages.

I tried deferring the message through the receiver object, replacing the processor object by the receiver object, but I always run into different dead ends.

I am using the Azure.Messaging.ServiceBus 7.10.0 lib for this.

Any ideas what I am missing - why do my session lock expire?

0 Answers
Related