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?