From c17b8ba576699047285b8f28291205202b1f40d5 Mon Sep 17 00:00:00 2001 From: Nabi Sobhi Date: Sun, 30 Aug 2026 20:39:01 +0200 Subject: [PATCH 1/2] Log an error when an exception bypasses the dead letter queue Consumer now accepts an optional ILogger (defaulting to NullLogger) and logs an error with the exception type, partition key and source topic when a registered bypass exception is about to propagate and crash the consumer. The real logger is resolved in both AddConsumer overloads. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- src/Dafda.Tests/Builders/ConsumerBuilder.cs | 11 ++++- src/Dafda.Tests/Consuming/TestConsumer.cs | 48 ++++++++++++++++++- src/Dafda.Tests/TestDoubles/LoggerSpy.cs | 43 +++++++++++++++++ .../ConsumerServiceCollectionExtensions.cs | 6 ++- src/Dafda/Consuming/Consumer.cs | 21 +++++++- 5 files changed, 122 insertions(+), 7 deletions(-) create mode 100644 src/Dafda.Tests/TestDoubles/LoggerSpy.cs diff --git a/src/Dafda.Tests/Builders/ConsumerBuilder.cs b/src/Dafda.Tests/Builders/ConsumerBuilder.cs index bedc6fee..364aa62c 100644 --- a/src/Dafda.Tests/Builders/ConsumerBuilder.cs +++ b/src/Dafda.Tests/Builders/ConsumerBuilder.cs @@ -3,6 +3,7 @@ using System; using Dafda.Consuming; using Dafda.Consuming.MessageFilters; +using Microsoft.Extensions.Logging; using TestDoubles; internal class ConsumerBuilder @@ -19,6 +20,7 @@ internal class ConsumerBuilder private int _maxRetries; private Func _deadLetterQueueBypass; private Func _retryBackoff; + private ILogger _logger; public ConsumerBuilder WithUnitOfWork(IHandlerUnitOfWork unitOfWork) { @@ -85,6 +87,12 @@ public ConsumerBuilder WithRetryBackoff(Func retryBackoff) return this; } + public ConsumerBuilder WithLogger(ILogger logger) + { + _logger = logger; + return this; + } + public Consumer Build() => new Consumer( _registry, @@ -97,5 +105,6 @@ public Consumer Build() => _deadLetterQueue, _maxRetries, _deadLetterQueueBypass, - _retryBackoff); + _retryBackoff, + _logger); } \ No newline at end of file diff --git a/src/Dafda.Tests/Consuming/TestConsumer.cs b/src/Dafda.Tests/Consuming/TestConsumer.cs index 3e2fe8fb..fff77afe 100644 --- a/src/Dafda.Tests/Consuming/TestConsumer.cs +++ b/src/Dafda.Tests/Consuming/TestConsumer.cs @@ -518,6 +518,48 @@ public async Task dead_letters_exceptions_that_do_not_match_the_bypass() Assert.Equal(1, deadLetterQueueSpy.SendCount); } + [Fact] + public async Task logs_error_when_exception_bypasses_the_dead_letter_queue() + { + var handler = new MessageHandlerSpy(() => throw new InvalidOperationException("fatal")); + + var loggerSpy = new LoggerSpy(); + + var sut = BuildConsumerWithHandler( + handler, + deadLetterQueue: new DeadLetterQueueSpy(), + deadLetterQueueBypass: exception => exception is InvalidOperationException, + logger: loggerSpy); + + await Assert.ThrowsAsync( + () => sut.ConsumeSingle(CancellationToken.None)); + + var logEntry = Assert.Single(loggerSpy.LogEntries); + Assert.Equal(LogLevel.Error, logEntry.LogLevel); + Assert.IsType(logEntry.Exception); + Assert.Equal( + "Exception of type System.InvalidOperationException bypassed the dead letter queue for message with key (null) from topic topic. Failing the consumer", + logEntry.Message); + } + + [Fact] + public async Task does_not_log_bypass_error_when_message_is_dead_lettered() + { + var handler = new MessageHandlerSpy(() => throw new InvalidOperationException("boom")); + + var loggerSpy = new LoggerSpy(); + + var sut = BuildConsumerWithHandler( + handler, + deadLetterQueue: new DeadLetterQueueSpy(), + deadLetterQueueBypass: exception => exception is FormatException, + logger: loggerSpy); + + await sut.ConsumeSingle(CancellationToken.None); + + Assert.Empty(loggerSpy.LogEntries); + } + [Fact] public async Task does_not_delay_between_retries_when_no_backoff_is_configured() { @@ -626,7 +668,8 @@ private static Consumer BuildConsumerWithHandler( IDeadLetterQueue deadLetterQueue = null, int maxRetries = 0, Func deadLetterQueueBypass = null, - Func retryBackoff = null) + Func retryBackoff = null, + ILogger logger = null) { var registration = new MessageRegistrationBuilder() .WithHandlerInstanceType(handler.GetType()) @@ -650,7 +693,8 @@ private static Consumer BuildConsumerWithHandler( .WithMessageHandlerRegistry(registry) .WithMaxRetries(maxRetries) .WithDeadLetterQueueBypass(deadLetterQueueBypass) - .WithRetryBackoff(retryBackoff); + .WithRetryBackoff(retryBackoff) + .WithLogger(logger); if (deadLetterQueue != null) { diff --git a/src/Dafda.Tests/TestDoubles/LoggerSpy.cs b/src/Dafda.Tests/TestDoubles/LoggerSpy.cs new file mode 100644 index 00000000..a0a135f4 --- /dev/null +++ b/src/Dafda.Tests/TestDoubles/LoggerSpy.cs @@ -0,0 +1,43 @@ +using System; +using System.Collections.Generic; +using Microsoft.Extensions.Logging; + +namespace Dafda.Tests.TestDoubles +{ + internal class LoggerSpy : ILogger + { + public IList LogEntries { get; } = new List(); + + public void Log(LogLevel logLevel, EventId eventId, TState state, Exception exception, Func formatter) + { + LogEntries.Add(new LogEntry(logLevel, formatter(state, exception), exception)); + } + + public bool IsEnabled(LogLevel logLevel) => true; + + public IDisposable BeginScope(TState state) => NullScope.Instance; + + internal class LogEntry + { + public LogEntry(LogLevel logLevel, string message, Exception exception) + { + LogLevel = logLevel; + Message = message; + Exception = exception; + } + + public LogLevel LogLevel { get; } + public string Message { get; } + public Exception Exception { get; } + } + + private class NullScope : IDisposable + { + public static readonly NullScope Instance = new NullScope(); + + public void Dispose() + { + } + } + } +} diff --git a/src/Dafda/Configuration/ConsumerServiceCollectionExtensions.cs b/src/Dafda/Configuration/ConsumerServiceCollectionExtensions.cs index f700ad31..7eddf0b2 100644 --- a/src/Dafda/Configuration/ConsumerServiceCollectionExtensions.cs +++ b/src/Dafda/Configuration/ConsumerServiceCollectionExtensions.cs @@ -48,7 +48,8 @@ public static void AddConsumer(this IServiceCollection services, Action>() ), configuration.GroupId, configuration.ConsumerErrorHandler @@ -86,7 +87,8 @@ public static void AddConsumer(this IServiceCollection services, Func>() ), configuration.GroupId, configuration.ConsumerErrorHandler diff --git a/src/Dafda/Consuming/Consumer.cs b/src/Dafda/Consuming/Consumer.cs index ff8acc54..4f55fa12 100644 --- a/src/Dafda/Consuming/Consumer.cs +++ b/src/Dafda/Consuming/Consumer.cs @@ -6,6 +6,8 @@ namespace Dafda.Consuming; using Diagnostics; using Interfaces; using MessageFilters; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; internal class Consumer( MessageHandlerRegistry messageHandlerRegistry, @@ -18,7 +20,8 @@ internal class Consumer( IDeadLetterQueue deadLetterQueue = null, int maxRetries = 0, Func deadLetterQueueBypass = null, - Func retryBackoff = null) + Func retryBackoff = null, + ILogger logger = null) : IConsumer, IDisposable { private readonly LocalMessageDispatcher _localMessageDispatcher = new( @@ -29,6 +32,8 @@ internal class Consumer( private readonly IDeadLetterQueue _deadLetterQueue = deadLetterQueue ?? NullDeadLetterQueue.Instance; + private readonly ILogger _logger = logger ?? NullLogger.Instance; + public async Task ConsumeAll(CancellationToken cancellationToken) { using var consumerScope = consumerScopeFactory.CreateConsumerScope(); @@ -72,8 +77,20 @@ private async Task Dispatch(MessageResult messageResult, CancellationToken cance await _localMessageDispatcher.Dispatch(messageResult, cancellationToken); return; } - catch (Exception exception) when (deadLetterQueueEnabled && !cancellationToken.IsCancellationRequested && !ShouldBypassDeadLetterQueue(exception)) + catch (Exception exception) when (deadLetterQueueEnabled && !cancellationToken.IsCancellationRequested) { + if (ShouldBypassDeadLetterQueue(exception)) + { + _logger.LogError( + exception, + "Exception of type {ExceptionType} bypassed the dead letter queue for message with key {Key} from topic {SourceTopic}. Failing the consumer", + exception.GetType().FullName, + messageResult.PartitionKey, + messageResult.Topic); + + throw; + } + if (attempt < maxRetries) { attempt++; From 3c230aa40b72b4658fb13477f18ef27bf59fa047 Mon Sep 17 00:00:00 2001 From: Nabi Sobhi Date: Thu, 17 Sep 2026 14:07:11 +0200 Subject: [PATCH 2/2] Preserve the original exception when a bypass predicate throws Evaluating the bypass predicate outside an exception filter meant a throwing predicate replaced the original handler exception. ShouldBypassDeadLetterQueue now treats a failing predicate as a bypass, so the bare rethrow propagates the original exception instead of dead-lettering it or losing it. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- src/Dafda.Tests/Consuming/TestConsumer.cs | 20 ++++++++++++++++++++ src/Dafda/Consuming/Consumer.cs | 19 ++++++++++++++++++- 2 files changed, 38 insertions(+), 1 deletion(-) diff --git a/src/Dafda.Tests/Consuming/TestConsumer.cs b/src/Dafda.Tests/Consuming/TestConsumer.cs index fff77afe..8a7df103 100644 --- a/src/Dafda.Tests/Consuming/TestConsumer.cs +++ b/src/Dafda.Tests/Consuming/TestConsumer.cs @@ -560,6 +560,26 @@ public async Task does_not_log_bypass_error_when_message_is_dead_lettered() Assert.Empty(loggerSpy.LogEntries); } + [Fact] + public async Task propagates_the_original_exception_when_the_bypass_predicate_throws() + { + var handler = new MessageHandlerSpy(() => throw new InvalidOperationException("original")); + + var deadLetterQueueSpy = new DeadLetterQueueSpy(); + + var sut = BuildConsumerWithHandler( + handler, + deadLetterQueue: deadLetterQueueSpy, + maxRetries: 3, + deadLetterQueueBypass: _ => throw new FormatException("predicate blew up")); + + var exception = await Assert.ThrowsAsync( + () => sut.ConsumeSingle(CancellationToken.None)); + + Assert.Equal("original", exception.Message); + Assert.Equal(0, deadLetterQueueSpy.SendCount); + } + [Fact] public async Task does_not_delay_between_retries_when_no_backoff_is_configured() { diff --git a/src/Dafda/Consuming/Consumer.cs b/src/Dafda/Consuming/Consumer.cs index 4f55fa12..3f723ae2 100644 --- a/src/Dafda/Consuming/Consumer.cs +++ b/src/Dafda/Consuming/Consumer.cs @@ -115,9 +115,26 @@ private TimeSpan GetRetryDelay(int attempt) return retryBackoff == null ? TimeSpan.Zero : retryBackoff(attempt); } + /// + /// Determines whether should bypass the dead letter queue. + /// A predicate that itself throws is treated as a bypass, so the original exception + /// propagates rather than being replaced by the predicate's exception or dead-lettered. + /// private bool ShouldBypassDeadLetterQueue(Exception exception) { - return deadLetterQueueBypass != null && deadLetterQueueBypass(exception); + if (deadLetterQueueBypass == null) + { + return false; + } + + try + { + return deadLetterQueueBypass(exception); + } + catch + { + return true; + } } public void Dispose()