Question 242115 in Azure Question posted in
The official documentation to get you started can be found here.

How to implement the retry pattern when processing a message from a queue or topic subscription in Azure Service Bus fails due to transient failure

I haven’t found a clear and comprehensive explanation as to how to implement the retry pattern when trying to process a message peeked from an Azure Service Bus. Processing of message is impeded by a transient failure (i.e. a server or service is temporarily unavailable or busy) and message needs to be peeked and its processing tried again after a specific delay.

Tags:

2

Answers


  1. Chosen as BEST ANSWER
    0

    The Retry pattern can be used to try and process a message again after a transient failure impedes the message processing (i.e. a service or server required is busy). The retry action needs to be delayed to give the systems or services that have the failure time to possibly recover. After a defined number of attempts it should also be possible to just give up and send the message to a Dead Letter Queue for someone to re-enqueue it manually once failure has been corrected.

    Possible ways to implement this pattern for messaging processing:

    1. Abandoning the message

    When configuring the queue or topic subscription, you can set the Message Lock Duration for the delay wanted for retrying, and the Max delivery count to the number of attempts desired. When a transient failure occurs, you can just finish the processing without completing or explicitly abandoning the message. Azure will remove the lock and make the message available again for consumption only after the Message Lock Duration has passed, hence causing the delay expected before the message processing is retried. This approach allows only a constant delay (for exponential backoff see options 2 and 3).

    Pros:

    • Simplest approach
    • Works well for both queues and topics with one or multiple subscriptions (message is returned to the queue, or to the specific subscription).

    Cons:

    • It does not allow delays with exponential backoff.

    2. Use the ScheduledEnqueueTimeUtc property

    You can clone the original message and set the ScheduledEnqueueTimeUTC property with a value that represent the delay desired, i.e. current date and time plus the delay (you will need to add a message custom property or a field to the message body to keep an attempt counter, that you can increase after each retry; you can use this counter to calculate the delay as exponential backoff, and also to check the maximum number of attempts after which you can send the message to the Dead Letter Queue). Publish the new message to the same queue/topic and complete (remove) the old one. The message will be available again after the delay has passed. In the case of topics it will work when there is only one subscription (topics with multiple subscription might have the problem of unnecessarily enqueueing a message to other subscriptions that actually succeeded to process the message before).

    Pros:

    • Allows exponential backoff.

    Cons:

    • It does not work well for topics with multiple subscriptions (message is send again to all the subscriptions, not just the one that triggered the retry action).

    3. Configure a queue specific for the messages to retry

    You can clone the original message and invoke the ScheduleMessageAsync method passing the new message and the value in seconds for the delay desired (you will need to add a message custom property or a field to the message body to keep an attempt counter, that you can increase after each retry; you can use this counter to calculate the delay as exponential backoff, and also to check the maximum number of attempts after which you can send the message to the Dead Letter Queue). Publish the new message to the queue specified for the messages to retry, and complete (remove) the old one. Your client will need to subscribe to both the original topic subscription (for the normal processing) and also to the queue you defined for the messages to retry (for retry processing). From the retry queue, the message will be available again after the delay has passed.

    Pros:

    • Allows exponential backoff.
    • It work well for topics with multiple subscriptions (message is send again to all the subscriptions, not just the one that triggered the retry action).

    Cons:

    • Overkill for queue and topics with just one subscription.

  2. 0

    Here’s a step-by-step guide to help you implement the retry pattern effectively:

    Wrap your message processing logic inside a try-catch block to catch exceptions that may occur during message processing.

    from azure.servicebus import ServiceBusClient, ServiceBusMessage

def process_message(message):
    try:
        # Your message processing logic here
        print(f"Processing message: {message.body}")
    except Exception as e:
        # Handle exceptions that occurred during processing
        print(f"Error processing message: {e}")

    Azure Service Bus provides built-in retry policies. You can configure these policies to automatically handle transient errors (like network issues, timeouts) during message processing.

    from azure.servicebus import ServiceBusClient, ServiceBusMessage
from azure.servicebus.common import RetriesExhausted

# Create a Service Bus client with retry options
retry_options = ServiceBusClientRetryOptions(max_retries=3)
servicebus_client = ServiceBusClient.from_connection_string(conn_str, retry_options=retry_options)

with servicebus_client:
    receiver = servicebus_client.get_queue_receiver(queue_name="myqueue")

    try:
        for msg in receiver:
            process_message(msg)
            msg.complete()
    except RetriesExhausted as e:
        print(f"Failed to process message after retries: {e}")

    If the built-in retry options are not sufficient for your use case, you can implement your own retry logic using a loop. Here’s an example of manual retry with a delay between retries:

    from time import sleep

max_retries = 3
retry_delay_seconds = 5

def process_message_with_retry(message):
    retries = 0
    while retries < max_retries:
        try:
            process_message(message)
            message.complete()
            break  # Exit the loop if processing is successful
        except Exception as e:
            print(f"Error processing message: {e}")
            retries += 1
            if retries < max_retries:
                print(f"Retrying in {retry_delay_seconds} seconds...")
                sleep(retry_delay_seconds)
            else:
                print("Max retries reached. Message processing failed.")

with servicebus_client:
    receiver = servicebus_client.get_queue_receiver(queue_name="myqueue")

    for msg in receiver:
        process_message_with_retry(msg)

    Remember that the retry pattern is designed to handle transient errors. If an error is not transient and the message processing will likely fail on each retry attempt, it’s important to handle such scenarios differently (e.g., moving the message to an error queue for manual inspection).

    Login or Signup to reply.
Please signup or login to give your own answer.

Back To Top
Search