EIP – Idempotent Receivers

Post by: syed hussain in All Integration Design Patterns

Summary

An idempotent receiver is a message receiver that ensures that the processing of a message is executed only once, even if the message is received multiple times. This is useful when there is a possibility of messages being duplicated due to network failures or other issues. In Azure, you can implement an idempotent receiver using Azure Functions and Azure Service Bus.

Implementation

Here’s how you can create an idempotent receiver in Azure Functions and Azure Service Bus:

  1. Create an Azure Function with a Service Bus trigger. In the Azure portal, go to the Functions app and create a new function. Choose the Service Bus trigger and select the Service Bus namespace and the queue or topic you want to receive messages from.
  2. Enable the “Defer” feature in the Service Bus trigger. This feature allows you to defer the processing of a message until you are ready to handle it. To enable this feature, set the “IsSessionsEnabled” property of the Service Bus trigger to true.
  3. Use the message SessionId to ensure idempotency. In the function code, retrieve the SessionId of the message from the ServiceBusReceivedMessage object. The SessionId uniquely identifies the message and can be used to ensure that the message is processed only once.
  4. Store the SessionId in a persistent storage. To ensure that the function can track which messages have already been processed, store the SessionId in a persistent storage such as Azure Table Storage or Azure Cosmos DB.
  5. Check if the SessionId has already been processed. Before processing a message, check if the SessionId has already been processed by querying the persistent storage. If the SessionId exists, skip the processing of the message. If it does not exist, proceed with processing the message and store the SessionId in the persistent storage.

Here’s an example code for an idempotent receiver in Azure Functions and Azure Service Bus using C#:

using Microsoft.Azure.Functions.Worker;
using Microsoft.Azure.Functions.Worker.ServiceBus;
using Microsoft.Extensions.Logging;
using System.Threading.Tasks;

public static class IdempotentReceiver
{
    [Function("IdempotentReceiver")]
    public static async Task Run(
        [ServiceBusTrigger("myqueue", Connection = "ServiceBusConnection")] 
        ServiceBusReceivedMessage message,
        [ServiceBus("myqueue", Connection = "ServiceBusConnection")] 
        IAsyncCollector<ServiceBusMessage> collector,
        [Table("ProcessedMessages", Connection = "StorageConnection")] 
        IAsyncCollector<ProcessedMessageEntity> processedMessageTable,
        ILogger log)
    {
        log.LogInformation($"C# ServiceBus queue trigger function processed message: {message.MessageId}");

        var sessionId = message.SessionId;
        var processedMessage = await processedMessageTable.ReadAsync(sessionId);

        if (processedMessage == null)
        {
            // Process the message
            log.LogInformation($"Processing message {message.MessageId}");
            // ...
            // Send the output message using the Service Bus output binding
            await collector.AddAsync(new ServiceBusMessage(/* output message */));

            // Store the SessionId in the processed messages table
            var entity = new ProcessedMessageEntity(sessionId);
            await processedMessageTable.AddAsync(entity);
        }
        else
        {
            // The message has already been processed
            log.LogInformation($"Skipping message {message.MessageId}, already processed");
        }
    }
}

public class ProcessedMessageEntity
{
    public string PartitionKey { get; set; } = "ProcessedMessages";
    public string RowKey { get; set; }
    public ProcessedMessageEntity() { }
    public ProcessedMessageEntity(string sessionId) => RowKey = sessionId;
}

In this example, the function receives a message from the “myqueue” queue and retrieves

the SessionId from the ServiceBusReceivedMessage object. It then queries the “ProcessedMessages” table storage to check if the SessionId already exists. If the SessionId does not exist, the function processes the message and sends an output message using the Service Bus output binding. It then stores the SessionId in the “ProcessedMessages” table storage. If the SessionId already exists, the function skips the processing of the message.

Tags:
23 Apr 2023
0