Refactors Kafka consumer and adds features

This commit refactors the Kafka consumer implementation
for improved modularity and testability.

It introduces a configuration object for Kafka consumers,
centralizing configuration settings.

Also adds handlers, offset management and report interval functionality.

Author: pmdevers

examples/Examples/Aggregate/TestAggregate.cs

@@ -1,14 +1,18 @@
-using Microsoft.AspNetCore.Mvc.TagHelpers.Cache;
-
-namespace Examples.Aggregate;
+namespace Examples.Aggregate;
 
 public class TestCommands : IAggregateCommands<Guid>
 {
     public Guid Id { get; init; } = Guid.NewGuid();
     public required int Version { get; init; }
     public required string CommandName { get; init; }
 
+    public SetCounter? SetCounter { get; set; }
+
 }
+
+public record SetCounter(int Counter);
+
+
 public record Test : IAggregate<Guid, Test, TestCommands>
 {
     public Guid Id { get; init; }
@@ -25,6 +29,7 @@ public static Result<Test> Apply(Test state, TestCommands command)
             nameof(Create) => Create(command),
             nameof(Increment) => state.Increment(),
             nameof(Decrement) => state.Decrement(),
+            nameof(SetCounter) => state.SetCounter(command.SetCounter!),
             _ => Result.Failed(state, "Unknown command: " + command.CommandName)
         };
 
@@ -61,4 +66,22 @@ public Result<Test> Decrement()
             Counter = Counter - 1
         };
     }
+
+    public Result<Test> SetCounter(SetCounter cmd) 
+    {
+        if(cmd.Counter < 0)
+        {
+            return Result.Failed(this, "Counter cannot be less than 0.");
+        }
+
+        if(cmd.Counter > 100)
+        {
+            return Result.Failed(this, "Counter connot be more then 100.");
+        }
+
+        return this with
+        {
+            Counter = cmd.Counter
+        };
+    }
 }

examples/Examples/Program.cs

@@ -1,3 +1,4 @@
+using Confluent.Kafka;
 using Examples;
 using Examples.Aggregate;
 using MinimalKafka;
@@ -9,12 +10,12 @@
  {
      config
            .WithConfiguration(builder.Configuration.GetSection("Kafka"))
-           //.WithBootstrapServers("nas:9092")
+           .WithBootstrapServers("nas:9092")
            .WithGroupId(AppDomain.CurrentDomain.FriendlyName)
            .WithClientId(AppDomain.CurrentDomain.FriendlyName)
            .WithTransactionalId(AppDomain.CurrentDomain.FriendlyName)
-           //.WithOffsetReset(AutoOffsetReset.Earliest)
-           //.WithPartitionAssignedHandler((_, p) => p.Select(tp => new TopicPartitionOffset(tp, Offset.Beginning)))
+           .WithOffsetReset(AutoOffsetReset.Earliest)
+           .WithPartitionAssignedHandler((_, p) => p.Select(tp => new TopicPartitionOffset(tp, Offset.Beginning)))
            .WithJsonSerializers()
            .UseRocksDB(x =>
            {

src/MinimalKafka/Builders/KafkaConsumerBuilder.cs

@@ -1,9 +1,6 @@
 using Confluent.Kafka;
 using Microsoft.Extensions.DependencyInjection;
-using Microsoft.Extensions.Logging;
 using MinimalKafka.Internals;
-using MinimalKafka.Metadata;
-using MinimalKafka.Metadata.Internals;
 
 namespace MinimalKafka.Builders;
 
@@ -13,55 +10,15 @@ internal class KafkaConsumerBuilder(KafkaConsumerKey key, IKafkaBuilder builder)
 
     public IKafkaConsumer Build()
     {
-        var consumer = CreateConsumer();
-        var producer = builder.ServiceProvider.GetRequiredService<IKafkaProducer>();
-        var logger = builder.ServiceProvider.GetRequiredService<ILogger<KafkaConsumer>>();
-        var formatter = builder.ServiceProvider.GetRequiredService<KafkaTopicFormatter>();
-
-        return new KafkaConsumer(
-            key,
-            true,
-            consumer,
-            producer,
-            [.. Delegates],
-            formatter,
-            [.. builder.MetaData],
+        var conf = KafkaConsumerConfig.Create(key, Delegates, builder.MetaData);
+        
+        return ActivatorUtilities.CreateInstance<KafkaConsumer>(
             builder.ServiceProvider,
-            logger
+            conf
         );
     }
 
-    private IConsumer<byte[], byte[]> CreateConsumer()
-    {
-        var config = builder.MetaData.OfType<IConfigMetadata>().First();
-        var handlers = builder.MetaData.OfType<IConsumerHandlerMetadata>().FirstOrDefault() ??
-            new ConsumerHandlerMetadata();
-
-        return new ConsumerBuilder<byte[], byte[]>(config.ConsumerConfig.AsEnumerable())
-            .SetKeyDeserializer(Deserializers.ByteArray)
-            .SetValueDeserializer(Deserializers.ByteArray)
-            .SetStatisticsHandler(handlers?.StatisticsHandler)
-            .SetErrorHandler(handlers?.ErrorHandler)
-            .SetLogHandler(handlers?.LogHandler)
-            .SetPartitionsAssignedHandler(handlers?.PartitionsAssignedHandler)
-            .SetPartitionsLostHandler(handlers?.PartitionsLostHandler)
-            .SetPartitionsRevokedHandler((c, partitions) =>
-            {
-                Console.WriteLine($"Partitions revoked: [{string.Join(", ", partitions)}]");
-                try
-                {
-                    handlers?.PartitionsRevokedHandler?.Invoke(c, partitions);
-
-                    //c.Commit();  // commit offsets before losing partitions
-                    Console.WriteLine("Offsets committed on revoke.");
-                }
-                catch (KafkaException e) when (e.Error.Code == ErrorCode.Local_State)
-                {
-                    Console.WriteLine($"Skip commit on revoke: {e.Error.Reason}");
-                }
-            })
-            .Build();
-    }
+    
 
     internal void AddDelegate(KafkaDelegate del)
     {

src/MinimalKafka/Internals/KafkaConsumer.cs

@@ -5,22 +5,38 @@
 
 namespace MinimalKafka.Internals;
 
+internal class KafkaConsumerConfig
+{
+    public required KafkaConsumerKey Key { get; init; }
+    public required IReadOnlyList<KafkaDelegate> Delegates { get; init; }
+    public required IReadOnlyList<object> Metadata { get; init; }
+
+    internal static KafkaConsumerConfig Create(KafkaConsumerKey key, List<KafkaDelegate> delegates, List<object> metaData)
+        => new()
+        {
+            Key = key,
+            Delegates = [.. delegates],
+            Metadata = [.. metaData]
+        };
+}
+
 internal class KafkaConsumer(
-    KafkaConsumerKey consumerKey,
-    bool autoCommitEnabled,
-    IConsumer<byte[], byte[]> consumer,
+    KafkaConsumerConfig config,
     IKafkaProducer producer,
-    KafkaDelegate[] kafkaDelegates,
     KafkaTopicFormatter topicFormatter,
-    IReadOnlyList<object> metadata,
     IServiceProvider serviceProvider,
     ILogger<KafkaConsumer> logger) : IKafkaConsumer
 {
+    private long _recordsConsumed;
+    private readonly int _reportInterval = config.Metadata.ReportInterval();
+    private readonly bool _autoCommitEnabled = config.Metadata.AutoCommitEnabled();
+    private readonly IConsumer<byte[], byte[]> _consumer = CreateConsumer(config.Metadata);
+
     public void Subscribe()
     {
-        var topic = topicFormatter(consumerKey.TopicName);
-        consumer.Subscribe(topic);
-        logger.Subscribed(consumerKey.GroupId, consumerKey.ClientId, topic);
+        var topic = topicFormatter(config.Key.TopicName);
+        _consumer.Subscribe(topic);
+        logger.Subscribed(config.Key.GroupId, config.Key.ClientId, topic);
     }
 
     public async Task Consume(CancellationToken cancellationToken)
@@ -29,15 +45,20 @@ public async Task Consume(CancellationToken cancellationToken)
         {
             await using var scope = serviceProvider.CreateAsyncScope();
 
-            var result = consumer.Consume(cancellationToken);
+            var result = _consumer.Consume(cancellationToken);
 
-            var context = KafkaContext.Create(consumerKey, result.Message, scope.ServiceProvider, metadata);
+            if (++_recordsConsumed % _reportInterval == 0)
+            {
+                logger.RecordsConsumed(config.Key.GroupId, config.Key.ClientId, _recordsConsumed, result.Topic);
+            }
+
+            var context = KafkaContext.Create(config, result.Message, scope.ServiceProvider);
 
             var store = context.GetTopicStore();
 
             await store.AddOrUpdate(context.Key, context.Value);
 
-            foreach (var kafkaDelegate in kafkaDelegates)
+            foreach (var kafkaDelegate in config.Delegates)
             {
                 await kafkaDelegate.Invoke(context);
             }
@@ -50,37 +71,54 @@ public async Task Consume(CancellationToken cancellationToken)
         when(ex.CancellationToken == cancellationToken)
         {
 
-            logger.OperatonCanceled(consumerKey.GroupId, consumerKey.ClientId);
+            logger.OperatonCanceled(config.Key.GroupId, config.Key.ClientId);
         }
     }
 
     public void Close()
     {
         if (_isClosed)
         {
-            logger.ConsumerAlreadyClosed(consumerKey.GroupId, consumerKey.ClientId);
+            logger.ConsumerAlreadyClosed(config.Key.GroupId, config.Key.ClientId);
             return;
         }
 
         _isClosed = true;
 
-        consumer.Close();
-        consumer.Dispose();
-        logger.ConsumerClosed(consumerKey.GroupId, consumerKey.ClientId);
+        _consumer.Close();
+        _consumer.Dispose();
+        logger.ConsumerClosed(config.Key.GroupId, config.Key.ClientId);
     }
 
     private bool _isClosed;
 
     private void Commit(ConsumeResult<byte[], byte[]> result)
     {
-        if (!autoCommitEnabled)
+        if (!_autoCommitEnabled)
         {
-            logger.Committing(consumerKey.GroupId, consumerKey.ClientId);
+            logger.Committing(config.Key.GroupId, config.Key.ClientId);
 
-            consumer.StoreOffset(result);
-            consumer.Commit();
+            _consumer.StoreOffset(result);
+            _consumer.Commit();
         }
     }
+
+    private static IConsumer<byte[], byte[]> CreateConsumer(IReadOnlyList<object> metadata)
+    {
+        var config = metadata.ConsumerConfig();
+        var handlers = metadata.ConsumerHandlers();
+
+        return new ConsumerBuilder<byte[], byte[]>(config)
+            .SetKeyDeserializer(Deserializers.ByteArray)
+            .SetValueDeserializer(Deserializers.ByteArray)
+            .SetStatisticsHandler(handlers?.StatisticsHandler)
+            .SetErrorHandler(handlers?.ErrorHandler)
+            .SetLogHandler(handlers?.LogHandler)
+            .SetPartitionsAssignedHandler(handlers?.PartitionsAssignedHandler)
+            .SetPartitionsLostHandler(handlers?.PartitionsLostHandler)
+            .SetPartitionsRevokedHandler(handlers?.PartitionsRevokedHandler)
+            .Build();
+    }
 }
 
 

src/MinimalKafka/Internals/KafkaContextProducer.cs

@@ -58,7 +58,8 @@ public async Task ProduceAsync(KafkaContext ctx, CancellationToken ct)
 
     public async Task ProduceAsync<TKey, TValue>(string topic, TKey key, TValue value, Dictionary<string, string>? header = null)
     {
-        var context = KafkaContext.Create(KafkaConsumerKey.Random(Guid.NewGuid().ToString()), new() { Key = [], Value = [] }, _serviceProvider, []);
+        var config = KafkaConsumerConfig.Create(KafkaConsumerKey.Random(Guid.NewGuid().ToString()), [], []);
+        var context = KafkaContext.Create(config, new() { Key = [], Value = [] }, _serviceProvider);
         await context.ProduceAsync(topic, key, value, header);
         await ProduceAsync(context, CancellationToken.None);
     }