Log an error when an exception bypasses the dead letter queue

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<Exception, bool> _deadLetterQueueBypass;
     private Func<int, TimeSpan> _retryBackoff;
+    private ILogger<Consumer> _logger;
 
     public ConsumerBuilder WithUnitOfWork(IHandlerUnitOfWork unitOfWork)
     {
@@ -85,6 +87,12 @@ public ConsumerBuilder WithRetryBackoff(Func<int, TimeSpan> retryBackoff)
         return this;
     }
 
+    public ConsumerBuilder WithLogger(ILogger<Consumer> logger)
+    {
+        _logger = logger;
+        return this;
+    }
+
     public Consumer Build() =>
         new Consumer(
             _registry,
@@ -97,5 +105,6 @@ public Consumer Build() =>
             _deadLetterQueue,
             _maxRetries,
             _deadLetterQueueBypass,
-            _retryBackoff);
+            _retryBackoff,
+            _logger);
 }

src/Dafda.Tests/Consuming/TestConsumer.cs

@@ -518,6 +518,68 @@ 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<FooMessage>(() => throw new InvalidOperationException("fatal"));
+
+        var loggerSpy = new LoggerSpy<Consumer>();
+
+        var sut = BuildConsumerWithHandler(
+            handler,
+            deadLetterQueue: new DeadLetterQueueSpy(),
+            deadLetterQueueBypass: exception => exception is InvalidOperationException,
+            logger: loggerSpy);
+
+        await Assert.ThrowsAsync<InvalidOperationException>(
+            () => sut.ConsumeSingle(CancellationToken.None));
+
+        var logEntry = Assert.Single(loggerSpy.LogEntries);
+        Assert.Equal(LogLevel.Error, logEntry.LogLevel);
+        Assert.IsType<InvalidOperationException>(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<FooMessage>(() => throw new InvalidOperationException("boom"));
+
+        var loggerSpy = new LoggerSpy<Consumer>();
+
+        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 propagates_the_original_exception_when_the_bypass_predicate_throws()
+    {
+        var handler = new MessageHandlerSpy<FooMessage>(() => 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<InvalidOperationException>(
+            () => 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()
     {
@@ -626,7 +688,8 @@ private static Consumer BuildConsumerWithHandler(
         IDeadLetterQueue deadLetterQueue = null,
         int maxRetries = 0,
         Func<Exception, bool> deadLetterQueueBypass = null,
-        Func<int, TimeSpan> retryBackoff = null)
+        Func<int, TimeSpan> retryBackoff = null,
+        ILogger<Consumer> logger = null)
     {
         var registration = new MessageRegistrationBuilder()
             .WithHandlerInstanceType(handler.GetType())
@@ -650,7 +713,8 @@ private static Consumer BuildConsumerWithHandler(
             .WithMessageHandlerRegistry(registry)
             .WithMaxRetries(maxRetries)
             .WithDeadLetterQueueBypass(deadLetterQueueBypass)
-            .WithRetryBackoff(retryBackoff);
+            .WithRetryBackoff(retryBackoff)
+            .WithLogger(logger);
 
         if (deadLetterQueue != null)
         {

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<T> : ILogger<T>
+    {
+        public IList<LogEntry> LogEntries { get; } = new List<LogEntry>();
+
+        public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Exception exception, Func<TState, Exception, string> formatter)
+        {
+            LogEntries.Add(new LogEntry(logLevel, formatter(state, exception), exception));
+        }
+
+        public bool IsEnabled(LogLevel logLevel) => true;
+
+        public IDisposable BeginScope<TState>(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()
+            {
+            }
+        }
+    }
+}

src/Dafda/Configuration/ConsumerServiceCollectionExtensions.cs

@@ -48,7 +48,8 @@ public static void AddConsumer(this IServiceCollection services, Action<Consumer
                 configuration.DeadLetterQueueFactory(provider),
                 configuration.MaxRetries,
                 configuration.DeadLetterQueueBypass,
-                configuration.RetryBackoff
+                configuration.RetryBackoff,
+                provider.GetRequiredService<ILogger<Consumer>>()
             ),
             configuration.GroupId,
             configuration.ConsumerErrorHandler
@@ -86,7 +87,8 @@ public static void AddConsumer(this IServiceCollection services, Func<IServicePr
                     configuration.DeadLetterQueueFactory(provider),
                     configuration.MaxRetries,
                     configuration.DeadLetterQueueBypass,
-                    configuration.RetryBackoff
+                    configuration.RetryBackoff,
+                    provider.GetRequiredService<ILogger<Consumer>>()
                 ),
                 configuration.GroupId,
                 configuration.ConsumerErrorHandler

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<Exception, bool> deadLetterQueueBypass = null,
-    Func<int, TimeSpan> retryBackoff = null)
+    Func<int, TimeSpan> retryBackoff = null,
+    ILogger<Consumer> 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<Consumer> _logger = logger ?? NullLogger<Consumer>.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++;
@@ -98,9 +115,26 @@ private TimeSpan GetRetryDelay(int attempt)
         return retryBackoff == null ? TimeSpan.Zero : retryBackoff(attempt);
     }
 
+    /// <summary>
+    /// Determines whether <paramref name="exception"/> 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.
+    /// </summary>
     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()