Variable quality of service on Kafka topics. Closes GH-602

JasperFx · Oct 20, 2023 · 1f500bc · 1f500bc
1 parent fcf79fd
commit 1f500bc
Show file tree

Hide file tree

Showing 3 changed files with 31 additions and 10 deletions.
diff --git a/src/Transports/Kafka/Wolverine.Kafka.Tests/compliance.cs b/src/Transports/Kafka/Wolverine.Kafka.Tests/compliance.cs
@@ -27,7 +27,7 @@ public async Task InitializeAsync()
 
         await SenderIs(opts =>
         {
-            opts.UseKafka("localhost:29092");
+            opts.UseKafka("localhost:29092").ConfigureConsumers(x => x.EnableAutoCommit = false);
 
             opts.ListenToKafkaTopic(senderTopic);
 

diff --git a/src/Transports/Kafka/Wolverine.Kafka/Internals/KafkaListener.cs b/src/Transports/Kafka/Wolverine.Kafka/Internals/KafkaListener.cs
@@ -14,6 +14,7 @@ internal class KafkaListener : IListener, IDisposable
     private readonly ConsumerConfig _config;
     private readonly IReceiver _receiver;
     private readonly string? _messageTypeName;
+    private readonly QualityOfService _qualityOfService;
 
     public KafkaListener(KafkaTopic topic, ConsumerConfig config, IReceiver receiver,
         ILogger<KafkaListener> logger)
@@ -27,14 +28,21 @@ public KafkaListener(KafkaTopic topic, ConsumerConfig config, IReceiver receiver
         _config = config;
         _receiver = receiver;
 
+        _qualityOfService = _config.EnableAutoCommit.HasValue && !_config.EnableAutoCommit.Value
+            ? QualityOfService.AtMostOnce
+            : QualityOfService.AtLeastOnce;
+
         _runner = Task.Run(async () =>
         {
             _consumer.Subscribe(topic.TopicName);
             try
             {
                 while (!_cancellation.IsCancellationRequested)
                 {
-                    // TODO -- watch that this isn't EnableAutoCommit = false
+                    if (_qualityOfService == QualityOfService.AtMostOnce)
+                    {
+                        _consumer.Commit();
+                    }
 
                     try
                     {
@@ -43,10 +51,13 @@ public KafkaListener(KafkaTopic topic, ConsumerConfig config, IReceiver receiver
 
                         var envelope = mapper.CreateEnvelope(result.Topic, message);
                         envelope.MessageType ??= _messageTypeName;
-
-                        await receiver.ReceivedAsync(this, envelope);
-
 
+
+                        await receiver.ReceivedAsync(this, envelope);
+                    }
+                    catch (OperationCanceledException)
+                    {
+                        // we're done here!
                     }
                     catch (Exception e)
                     {
@@ -67,11 +78,8 @@ public KafkaListener(KafkaTopic topic, ConsumerConfig config, IReceiver receiver
 
     public ValueTask CompleteAsync(Envelope envelope)
     {
-        if (_config.EnableAutoCommit != null)
-        {
-            _consumer.Commit();
-        }
-
+        // do nothing here, it's already ack'd before we get here
+
         return ValueTask.CompletedTask;
     }
 

diff --git a/src/Transports/Kafka/Wolverine.Kafka/KafkaTopic.cs b/src/Transports/Kafka/Wolverine.Kafka/KafkaTopic.cs
@@ -105,4 +105,17 @@ await client.CreateTopicsAsync(new[]
             throw;
         }
     }
+}
+
+public enum QualityOfService
+{
+    /// <summary>
+    /// "At least once" delivery guarantee by auto-ack'ing incoming messages
+    /// </summary>
+    AtLeastOnce,
+
+    /// <summary>
+    /// "At most once" delivery guarantee by trying to ack received messages before processing
+    /// </summary>
+    AtMostOnce
 }