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..8a7df103 100644 --- a/src/Dafda.Tests/Consuming/TestConsumer.cs +++ b/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(() => 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 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() { @@ -626,7 +688,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 +713,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..3f723ae2 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++; @@ -98,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()