Requeuing and moving erroneous messages to ASB DLQ

[pieperu]

Hello! I've been doing some debugging to work out why messages from an Azure Service Bus topic don't move to the DLQ if an exception is thrown, and instead the message is Acknowledged, and therefore not re-queued and removed from the queue.

I can see that in MessagePump there is a check for whether the exception is an DeferMessageAction exception, and if so, the message is re-queued and eventually when the count condition is met, it moves to the DLQ as expected.

Is this the best/only way to go about this re-queuing a message, to raise a DeferMessageAction exception?

If I've missed something, apologies - am still digging around now so will post back if I find anything else.

[DevJonny]

Is this the best/only way to go about this requeuing a message, to raise a DeferMessageAction execption?

Yes, using throw new DeferMessageAction() is meant for retrying a message that you're not ready to process yet. For example you've received it out of order.

[pieperu]

Thanks @DevJonny.

I was actually just trying to re-queued an event when an exception occurs in the command handler. Currently without catching and throwing a new DeferMessageAction, the event is acknowledged and removed from the topic. I wonder if i'm misusing this functionality and if i have a problem somewhere else. I will keep looking.

I've used this framework before with AWS (without ServiceActivator) and didn't need to explicitly do this. It's not an issue though, as i'm just throwing the DeferMessageAction in an attribute handler after I handle the original exception.

Thanks for the fast response.

[DevJonny]

Hmm yes requeuing on exception would be the behaviour that I would expect as well from using the RabbitMQ MessagingGateway. @preardon uses ASB in anger, he might be able to shed some more light on this case.

[preardon]

DeferMessageAction is meant for when you have transient faults (i.e. can't process at this time and want to process later)

When configuring your subscriptions you can set your Requeue count, when this count has been reached the message will be sent to the DLQ.

One of the things to note here is that the way that we Requeue. On ASB we take a copy of that message, put it on the end of the queue and ACK the original message, this is due to the functionality of the transport, I currently have a feature request open with the ASB team to have this added.

[pieperu]

Thanks @preardon.

I've actually got my retry count set to 0 so it should move to the DLQ one failure - I must have an issue in my configuration somewhere because I just cannot get this to work as expected (exception = DLQ):

new AzureServiceBusSubscription<BulkGiftingRequestedEvent>(
            new SubscriptionName(TopicCatalogue.BulkGiftingRequestedTopic),
            new ChannelName(ConsumerCatalogue.BulkGiftRequestConsumer),
            new RoutingKey(TopicCatalogue.BulkGiftingRequestedTopic),
            timeoutInMilliseconds: TimeSpan.FromSeconds(30).Milliseconds,
            makeChannels: OnMissingChannel.Create,
            requeueCount: 0, // We rely on the polly retry policy for this
            isAsync: true,
            noOfPerformers: 1),

[preardon]

I'll have a quick look at the code now and get back to you

[pieperu]

Thanks, for context, this is the relevant bootstrapping code:

services
    .AddServiceActivator(options =>
    {
        options.Subscriptions = _subscriptions;
        options.ChannelFactory = new AzureServiceBusChannelFactory(asbConsumerFactory);
        options.UseScoped = false;
    })
    .AutoFromAssemblies();

    services.AddHostedService<ServiceActivatorHostedService>();

    services.AddBrighter(configure =>
    {        
        configure.PolicyRegistry = new PolicyRegistry
        {
                            { CommandProcessor.RETRYPOLICYASYNC, PolicyCatalogue.DefaultRetryAsync },
                            { CommandProcessor.RETRYPOLICY, PolicyCatalogue.DefaultRetry },
                            { CommandProcessor.CIRCUITBREAKERASYNC, PolicyCatalogue.DefaultCircuitBreakerAsync },
                            { CommandProcessor.CIRCUITBREAKER, PolicyCatalogue.DefaultCircuitBreaker }
        };
    })
    .AutoFromAssemblies();

[pieperu]

And it could very well be because I’ve configured something incorrectly. Is what I’ve described above the expected behaviour (No requeue on unhandled exception without explicitly raising DeferAction)?

[iancooper]

So there are different kinds of queue here:

• Dead-Letter (DLQ): Moving a message here means "could not deliver". (May be triggered by too many retries on some brokers).

Brokers tend to support a DLQ "out of the box".

The following queue is well-defined but usually not supported by a broker, so it is for the end-user/framework

• Invalid Message: Moving a message here means "could not deserialize" your message, usually configuration or version issues

It is worth us thinking about support for this, as it may still be a good message, we just can't consume it, or it was meant for someone else.

Some folks then wonder - what happens if I get an exception in my handler. There are two types of exception we can think about:

• Retryable. We fail because of a timing issue, something else is not available, or we are missing a required change to our state
• Non-retryable. No matter how many times we try we will not succeed. The message is "poison pill" if not removed.

For the former case we offer DeferMessageAction - please retry this later.

For the latter case, log your exception locally, and ack the message to delete it from the queue (or advance your offset). You cannot complete this work, it is an error, like any other, and your logs should be used to pick it up.

Now some folks like to create an "error queue" and report their exceptions there. If that is your thing go for it, but I'd rather use Kibana or the like to find them from a log. See https://12factor.net/logs

If you must have an error queue, just write some middleware (just a request handler) to intercept any exceptions that bubble out of the handler send it to your error queue, and add your corresponding attribute to your destination handler class.

[ExceptionsToErrorQueue] public Handle...

But DONT push these to the DLQ. That is not what it is for.

[iancooper]

BTW @pieperu we do publish details on how to write your own middleware, but reach out if it is not clear. But that is the solution you should use if you want to intercept our pipeline to create different behavior.

[iancooper]

It is possible we should have called DeferMessageAction, RetryMessageAction. We mean it in the sense of "do it again". But it's possible we have created confusion here

[pieperu]

Thanks so much for the clarification @iancooper, and thank you @preardon for all of your attention on this. This has cleared everything up for me. I really just wanted to know that I was using the DeferMessageAction as it was intended, or if I had something misconfigured somewhere.

I'm currently handling it exactly as you've suggested at the moment Ian with an attribute handler to ensure DeferMessageAction is raised at the appropriate time.

[RequestLoggingAsync(0, HandlerTiming.Before)]
[RequeueOnException(1)] <-- This one throws the DeferMessageAction in it's catch
[LogContext(2)]
[UsePolicyAsync(PolicyCatalogue.ActorPolicyKey, 3)]
[Monitor(4)]

[pieperu]

My question was essentially:

Is raising a DeferMessageAction the correct way to requeue an item, and eventually move it to the DLQ?

The short answer is yes.

Great explanation here in one of the comments in this discussion.