Skip to content

Commit 15834b6

Browse files
EtherZazarusz
EtherZa
authored and
zarusz
committed
#356 Abandon/dead letter message
Signed-off-by: Richard Pringle <[email protected]>
1 parent fd2e9e3 commit 15834b6

File tree

41 files changed

+428
-120
lines changed

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

41 files changed

+428
-120
lines changed

‎docs/intro.md

+45-3
Original file line numberDiff line numberDiff line change
@@ -1089,10 +1089,11 @@ The returned `ConsumerErrorHandlerResult` object is used to override the executi
10891089

10901090
| Result | Description |
10911091
| ------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
1092+
| Abandon | The message should be sent to the dead letter queue/exchange. **Not supported by all transports.** |
10921093
| Failure | The message failed to be processed and should be returned to the queue |
10931094
| Success | The pipeline must treat the message as having been processed successfully |
10941095
| SuccessWithResponse | The pipeline to treat the message as having been processed successfully, returning the response to the request/response invocation ([IRequestResponseBus<T>](../src/SlimMessageBus/RequestResponse/IRequestResponseBus.cs)) |
1095-
| Retry | Re-create and execute the pipeline (including the message scope when using [per-message DI container scopes](#per-message-di-container-scope)) |
1096+
| Retry | Re-create and execute the pipeline (including the message scope when using [per-message DI container scopes](#per-message-di-container-scope)) |
10961097

10971098
To enable SMB to recognize the error handler, it must be registered within the Microsoft Dependency Injection (MSDI) framework:
10981099

@@ -1119,15 +1120,47 @@ Transport plugins provide specialized error handling interfaces. Examples includ
11191120

11201121
This approach allows for transport-specific error handling, ensuring that specialized handlers can be prioritized.
11211122

1122-
Sample retry with exponential back-off (using the [ConsumerErrorHandler](../src/SlimMessageBus.Host/Consumer/ErrorHandling/ConsumerErrorHandler.cs) abstract implementation):
1123+
1124+
### Abandon
1125+
#### Azure Service Bus
1126+
The Azure Service Bus transport has full support for abandoning messages to the dead letter queue.
1127+
1128+
#### RabbitMQ
1129+
Abandon will issue a `Nack` with `requeue: false`.
1130+
1131+
#### Other transports
1132+
No other transports currently support `Abandon` and calling `Abandon` will result in `NotSupportedException` being thrown.
1133+
1134+
### Failure
1135+
#### RabbitMQ
1136+
While RabbitMQ supports dead letter exchanges, SMB's default implementation is not to requeue messages on `Failure`. If requeuing is required, it can be enabled by setting `RequeueOnFailure()` when configuring a consumer/handler.
1137+
1138+
Please be aware that as RabbitMQ does not have a maximum delivery count and enabling requeue may result in an infinite message loop. When `RequeueOnFailure()` has been set, it is the developer's responsibility to configure an appropriate `IConsumerErrorHandler` that will `Abandon` all non-transient exceptions.
11231139

1140+
```cs
1141+
.Handle<EchoRequest, EchoResponse>(x => x
1142+
.Queue("echo-request-handler")
1143+
.ExchangeBinding("test-echo")
1144+
.DeadLetterExchange("echo-request-handler-dlq")
1145+
// requeue a message on failure
1146+
.RequeueOnFailure()
1147+
.WithHandler<EchoRequestHandler>())
1148+
```
1149+
1150+
### Example usage
1151+
Retry with exponential back-off and short-curcuit dead letter on non-transient exceptions (using the [ConsumerErrorHandler](../src/SlimMessageBus.Host/Consumer/ErrorHandling/ConsumerErrorHandler.cs) abstract implementation):
11241152
```cs
11251153
public class RetryHandler<T> : ConsumerErrorHandler<T>
11261154
{
11271155
private static readonly Random _random = new();
11281156

11291157
public override async Task<ConsumerErrorHandlerResult> OnHandleError(T message, IConsumerContext consumerContext, Exception exception, int attempts)
11301158
{
1159+
if (!IsTranientException(exception))
1160+
{
1161+
return Abandon();
1162+
}
1163+
11311164
if (attempts < 3)
11321165
{
11331166
var delay = (attempts * 1000) + (_random.Next(1000) - 500);
@@ -1137,9 +1170,18 @@ public class RetryHandler<T> : ConsumerErrorHandler<T>
11371170

11381171
return Failure();
11391172
}
1173+
1174+
private static bool IsTransientException(Exception exception)
1175+
{
1176+
while (exception is not SqlException && exception.InnerException != null)
1177+
{
1178+
exception = exception.InnerException;
1179+
}
1180+
1181+
return exception is SqlException { Number: -2 or 1205 }; // Timeout or deadlock
1182+
}
11401183
}
11411184
```
1142-
11431185
## Logging
11441186

11451187
SlimMessageBus uses [Microsoft.Extensions.Logging.Abstractions](https://www.nuget.org/packages/Microsoft.Extensions.Logging.Abstractions):

‎docs/intro.t.md

+45-3
Original file line numberDiff line numberDiff line change
@@ -1067,10 +1067,11 @@ The returned `ConsumerErrorHandlerResult` object is used to override the executi
10671067

10681068
| Result | Description |
10691069
| ------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
1070+
| Abandon | The message should be sent to the dead letter queue/exchange. **Not supported by all transports.** |
10701071
| Failure | The message failed to be processed and should be returned to the queue |
10711072
| Success | The pipeline must treat the message as having been processed successfully |
10721073
| SuccessWithResponse | The pipeline to treat the message as having been processed successfully, returning the response to the request/response invocation ([IRequestResponseBus<T>](../src/SlimMessageBus/RequestResponse/IRequestResponseBus.cs)) |
1073-
| Retry | Re-create and execute the pipeline (including the message scope when using [per-message DI container scopes](#per-message-di-container-scope)) |
1074+
| Retry | Re-create and execute the pipeline (including the message scope when using [per-message DI container scopes](#per-message-di-container-scope)) |
10741075

10751076
To enable SMB to recognize the error handler, it must be registered within the Microsoft Dependency Injection (MSDI) framework:
10761077

@@ -1097,15 +1098,47 @@ Transport plugins provide specialized error handling interfaces. Examples includ
10971098

10981099
This approach allows for transport-specific error handling, ensuring that specialized handlers can be prioritized.
10991100

1100-
Sample retry with exponential back-off (using the [ConsumerErrorHandler](../src/SlimMessageBus.Host/Consumer/ErrorHandling/ConsumerErrorHandler.cs) abstract implementation):
1101+
1102+
### Abandon
1103+
#### Azure Service Bus
1104+
The Azure Service Bus transport has full support for abandoning messages to the dead letter queue.
1105+
1106+
#### RabbitMQ
1107+
Abandon will issue a `Nack` with `requeue: false`.
1108+
1109+
#### Other transports
1110+
No other transports currently support `Abandon` and calling `Abandon` will result in `NotSupportedException` being thrown.
1111+
1112+
### Failure
1113+
#### RabbitMQ
1114+
While RabbitMQ supports dead letter exchanges, SMB's default implementation is not to requeue messages on `Failure`. If requeuing is required, it can be enabled by setting `RequeueOnFailure()` when configuring a consumer/handler.
1115+
1116+
Please be aware that as RabbitMQ does not have a maximum delivery count and enabling requeue may result in an infinite message loop. When `RequeueOnFailure()` has been set, it is the developer's responsibility to configure an appropriate `IConsumerErrorHandler` that will `Abandon` all non-transient exceptions.
11011117

1118+
```cs
1119+
.Handle<EchoRequest, EchoResponse>(x => x
1120+
.Queue("echo-request-handler")
1121+
.ExchangeBinding("test-echo")
1122+
.DeadLetterExchange("echo-request-handler-dlq")
1123+
// requeue a message on failure
1124+
.RequeueOnFailure()
1125+
.WithHandler<EchoRequestHandler>())
1126+
```
1127+
1128+
### Example usage
1129+
Retry with exponential back-off and short-curcuit dead letter on non-transient exceptions (using the [ConsumerErrorHandler](../src/SlimMessageBus.Host/Consumer/ErrorHandling/ConsumerErrorHandler.cs) abstract implementation):
11021130
```cs
11031131
public class RetryHandler<T> : ConsumerErrorHandler<T>
11041132
{
11051133
private static readonly Random _random = new();
11061134

11071135
public override async Task<ConsumerErrorHandlerResult> OnHandleError(T message, IConsumerContext consumerContext, Exception exception, int attempts)
11081136
{
1137+
if (!IsTranientException(exception))
1138+
{
1139+
return Abandon();
1140+
}
1141+
11091142
if (attempts < 3)
11101143
{
11111144
var delay = (attempts * 1000) + (_random.Next(1000) - 500);
@@ -1115,9 +1148,18 @@ public class RetryHandler<T> : ConsumerErrorHandler<T>
11151148

11161149
return Failure();
11171150
}
1151+
1152+
private static bool IsTransientException(Exception exception)
1153+
{
1154+
while (exception is not SqlException && exception.InnerException != null)
1155+
{
1156+
exception = exception.InnerException;
1157+
}
1158+
1159+
return exception is SqlException { Number: -2 or 1205 }; // Timeout or deadlock
1160+
}
11181161
}
11191162
```
1120-
11211163
## Logging
11221164

11231165
SlimMessageBus uses [Microsoft.Extensions.Logging.Abstractions](https://www.nuget.org/packages/Microsoft.Extensions.Logging.Abstractions):

‎src/SlimMessageBus.Host.AmazonSQS/Consumer/ISqsConsumerErrorHandler.cs

+1-1
Original file line numberDiff line numberDiff line change
@@ -2,4 +2,4 @@
22

33
public interface ISqsConsumerErrorHandler<in T> : IConsumerErrorHandler<T>;
44

5-
public abstract class SqsConsumerErrorHandler<T> : ConsumerErrorHandler<T>;
5+
public abstract class SqsConsumerErrorHandler<T> : ConsumerErrorHandler<T>, ISqsConsumerErrorHandler<T>;

‎src/SlimMessageBus.Host.AmazonSQS/Consumer/SqsBaseConsumer.cs

+5
Original file line numberDiff line numberDiff line change
@@ -116,6 +116,11 @@ protected async Task Run()
116116
.ToDictionary(x => x.Key, x => HeaderSerializer.Deserialize(x.Key, x.Value));
117117

118118
var r = await MessageProcessor.ProcessMessage(message, messageHeaders, cancellationToken: CancellationToken).ConfigureAwait(false);
119+
if (r.Result == ProcessResult.Abandon)
120+
{
121+
throw new NotSupportedException("Transport does not support abandoning messages");
122+
}
123+
119124
if (r.Exception != null)
120125
{
121126
Logger.LogError(r.Exception, "Message processing error - Queue: {Queue}, MessageId: {MessageId}", Path, message.MessageId);

‎src/SlimMessageBus.Host.AzureEventHub/Consumer/EhPartitionConsumer.cs

+5
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,11 @@ public async Task ProcessEventAsync(ProcessEventArgs args)
4242

4343
var headers = GetHeadersFromTransportMessage(args.Data);
4444
var r = await MessageProcessor.ProcessMessage(args.Data, headers, cancellationToken: args.CancellationToken).ConfigureAwait(false);
45+
if (r.Result == ProcessResult.Abandon)
46+
{
47+
throw new NotSupportedException("Transport does not support abandoning messages");
48+
}
49+
4550
if (r.Exception != null)
4651
{
4752
// Note: The OnMessageFaulted was called at this point by the MessageProcessor.

‎src/SlimMessageBus.Host.AzureEventHub/Consumer/IEventHubConsumerErrorHandler.cs

+1-1
Original file line numberDiff line numberDiff line change
@@ -2,4 +2,4 @@
22

33
public interface IEventHubConsumerErrorHandler<in T> : IConsumerErrorHandler<T>;
44

5-
public abstract class EventHubConsumerErrorHandler<T> : ConsumerErrorHandler<T>;
5+
public abstract class EventHubConsumerErrorHandler<T> : ConsumerErrorHandler<T>, IEventHubConsumerErrorHandler<T>;

‎src/SlimMessageBus.Host.AzureServiceBus/Consumer/AsbBaseConsumer.cs

+34-20
Original file line numberDiff line numberDiff line change
@@ -137,18 +137,23 @@ private Task ServiceBusSessionProcessor_SessionClosingAsync(ProcessSessionEventA
137137
}
138138

139139
private Task ServiceBusSessionProcessor_ProcessMessageAsync(ProcessSessionMessageEventArgs args)
140-
=> ProcessMessageAsyncInternal(args.Message, args.CompleteMessageAsync, args.AbandonMessageAsync, args.CancellationToken);
140+
=> ProcessMessageAsyncInternal(args.Message, args.CompleteMessageAsync, args.AbandonMessageAsync, args.DeadLetterMessageAsync, args.CancellationToken);
141141

142142
private Task ServiceBusSessionProcessor_ProcessErrorAsync(ProcessErrorEventArgs args)
143143
=> ProcessErrorAsyncInternal(args.Exception, args.ErrorSource);
144144

145145
protected Task ServiceBusProcessor_ProcessMessagesAsync(ProcessMessageEventArgs args)
146-
=> ProcessMessageAsyncInternal(args.Message, args.CompleteMessageAsync, args.AbandonMessageAsync, args.CancellationToken);
146+
=> ProcessMessageAsyncInternal(args.Message, args.CompleteMessageAsync, args.AbandonMessageAsync, args.DeadLetterMessageAsync, args.CancellationToken);
147147

148148
protected Task ServiceBusProcessor_ProcessErrorAsync(ProcessErrorEventArgs args)
149149
=> ProcessErrorAsyncInternal(args.Exception, args.ErrorSource);
150150

151-
protected async Task ProcessMessageAsyncInternal(ServiceBusReceivedMessage message, Func<ServiceBusReceivedMessage, CancellationToken, Task> completeMessage, Func<ServiceBusReceivedMessage, IDictionary<string, object>, CancellationToken, Task> abandonMessage, CancellationToken token)
151+
protected async Task ProcessMessageAsyncInternal(
152+
ServiceBusReceivedMessage message,
153+
Func<ServiceBusReceivedMessage, CancellationToken, Task> completeMessage,
154+
Func<ServiceBusReceivedMessage, IDictionary<string, object>, CancellationToken, Task> abandonMessage,
155+
Func<ServiceBusReceivedMessage, string, string, CancellationToken, Task> deadLetterMessage,
156+
CancellationToken token)
152157
{
153158
// Process the message.
154159
Logger.LogDebug("Received message - Path: {Path}, SubscriptionName: {SubscriptionName}, SequenceNumber: {SequenceNumber}, DeliveryCount: {DeliveryCount}, MessageId: {MessageId}", TopicSubscription.Path, TopicSubscription.SubscriptionName, message.SequenceNumber, message.DeliveryCount, message.MessageId);
@@ -160,29 +165,38 @@ protected async Task ProcessMessageAsyncInternal(ServiceBusReceivedMessage messa
160165
// to avoid unnecessary exceptions.
161166
Logger.LogDebug("Abandon message - Path: {Path}, SubscriptionName: {SubscriptionName}, SequenceNumber: {SequenceNumber}, DeliveryCount: {DeliveryCount}, MessageId: {MessageId}", TopicSubscription.Path, TopicSubscription.SubscriptionName, message.SequenceNumber, message.DeliveryCount, message.MessageId);
162167
await abandonMessage(message, null, token).ConfigureAwait(false);
163-
164168
return;
165169
}
166170

167171
var r = await MessageProcessor.ProcessMessage(message, message.ApplicationProperties, cancellationToken: token).ConfigureAwait(false);
168-
if (r.Exception != null)
172+
switch (r.Result)
169173
{
170-
Logger.LogError(r.Exception, "Abandon message (exception occurred while processing) - Path: {Path}, SubscriptionName: {SubscriptionName}, SequenceNumber: {SequenceNumber}, DeliveryCount: {DeliveryCount}, MessageId: {MessageId}", TopicSubscription.Path, TopicSubscription.SubscriptionName, message.SequenceNumber, message.DeliveryCount, message.MessageId);
171-
172-
var messageProperties = new Dictionary<string, object>
173-
{
174-
// Set the exception message
175-
["SMB.Exception"] = r.Exception.Message
176-
};
177-
await abandonMessage(message, messageProperties, token).ConfigureAwait(false);
178-
179-
return;
174+
case ProcessResult.Success:
175+
// Complete the message so that it is not received again.
176+
// This can be done only if the subscriptionClient is created in ReceiveMode.PeekLock mode (which is the default).
177+
Logger.LogDebug("Complete message - Path: {Path}, SubscriptionName: {SubscriptionName}, SequenceNumber: {SequenceNumber}, DeliveryCount: {DeliveryCount}, MessageId: {MessageId}", TopicSubscription.Path, TopicSubscription.SubscriptionName, message.SequenceNumber, message.DeliveryCount, message.MessageId);
178+
await completeMessage(message, token).ConfigureAwait(false);
179+
return;
180+
181+
case ProcessResult.Abandon:
182+
Logger.LogError(r.Exception, "Dead letter message - Path: {Path}, SubscriptionName: {SubscriptionName}, SequenceNumber: {SequenceNumber}, DeliveryCount: {DeliveryCount}, MessageId: {MessageId}", TopicSubscription.Path, TopicSubscription.SubscriptionName, message.SequenceNumber, message.DeliveryCount, message.MessageId);
183+
await deadLetterMessage(message, r.Exception?.GetType().Name ?? string.Empty, r.Exception?.Message ?? string.Empty, token).ConfigureAwait(false);
184+
return;
185+
186+
case ProcessResult.Fail:
187+
var messageProperties = new Dictionary<string, object>();
188+
{
189+
// Set the exception message
190+
messageProperties.Add("SMB.Exception", r.Exception.Message);
191+
}
192+
193+
Logger.LogError(r.Exception, "Abandon message (exception occurred while processing) - Path: {Path}, SubscriptionName: {SubscriptionName}, SequenceNumber: {SequenceNumber}, DeliveryCount: {DeliveryCount}, MessageId: {MessageId}", TopicSubscription.Path, TopicSubscription.SubscriptionName, message.SequenceNumber, message.DeliveryCount, message.MessageId);
194+
await abandonMessage(message, messageProperties, token).ConfigureAwait(false);
195+
return;
196+
197+
default:
198+
throw new NotImplementedException();
180199
}
181-
182-
// Complete the message so that it is not received again.
183-
// This can be done only if the subscriptionClient is created in ReceiveMode.PeekLock mode (which is the default).
184-
Logger.LogDebug("Complete message - Path: {Path}, SubscriptionName: {SubscriptionName}, SequenceNumber: {SequenceNumber}, DeliveryCount: {DeliveryCount}, MessageId: {MessageId}", TopicSubscription.Path, TopicSubscription.SubscriptionName, message.SequenceNumber, message.DeliveryCount, message.MessageId);
185-
await completeMessage(message, token).ConfigureAwait(false);
186200
}
187201

188202
protected Task ProcessErrorAsyncInternal(Exception exception, ServiceBusErrorSource errorSource)

‎src/SlimMessageBus.Host.AzureServiceBus/Consumer/IServiceBusConsumerErrorHandler.cs

+1-1
Original file line numberDiff line numberDiff line change
@@ -2,4 +2,4 @@
22

33
public interface IServiceBusConsumerErrorHandler<in T> : IConsumerErrorHandler<T>;
44

5-
public abstract class ServiceBusConsumerErrorHandler<T> : ConsumerErrorHandler<T>;
5+
public abstract class ServiceBusConsumerErrorHandler<T> : ConsumerErrorHandler<T>, IServiceBusConsumerErrorHandler<T>;

‎src/SlimMessageBus.Host.Kafka/Consumer/IKafkaConsumerErrorHandler.cs

+1-1
Original file line numberDiff line numberDiff line change
@@ -2,4 +2,4 @@
22

33
public interface IKafkaConsumerErrorHandler<in T> : IConsumerErrorHandler<T>;
44

5-
public abstract class KafkaConsumerErrorHandler<T> : ConsumerErrorHandler<T>;
5+
public abstract class KafkaConsumerErrorHandler<T> : ConsumerErrorHandler<T>, IKafkaConsumerErrorHandler<T>;

‎src/SlimMessageBus.Host.Kafka/Consumer/KafkaPartitionConsumer.cs

+5
Original file line numberDiff line numberDiff line change
@@ -115,6 +115,11 @@ public async Task OnMessage(ConsumeResult message)
115115
}
116116

117117
var r = await _messageProcessor.ProcessMessage(message, messageHeaders, cancellationToken: _cancellationTokenSource.Token).ConfigureAwait(false);
118+
if (r.Result == ProcessResult.Abandon)
119+
{
120+
throw new NotSupportedException("Transport does not support abandoning messages");
121+
}
122+
118123
if (r.Exception != null)
119124
{
120125
// The IKafkaConsumerErrorHandler and OnMessageFaulted was called at this point by the MessageProcessor.

0 commit comments

Comments
 (0)