feat: enable workers count calculation in runtime

fix: change dependency injection lifetime management for worker and consumer/producer
fix: create AdminClient without authentication
perf: memory optimizations on MessageContext

samples/KafkaFlow.Sample.Dashboard/ConsumerLagWorkerBalancer.cs

@@ -0,0 +1,131 @@
+namespace KafkaFlow.Sample.Dashboard;
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+using Confluent.Kafka;
+using KafkaFlow.Clusters;
+using KafkaFlow.Configuration;
+using KafkaFlow.Consumers;
+using TopicMetadata = KafkaFlow.TopicMetadata;
+using TopicPartitionOffset = KafkaFlow.TopicPartitionOffset;
+
+public class ConsumerLagWorkerBalancer
+{
+    private readonly IClusterManager clusterManager;
+    private readonly IConsumerAccessor consumerAccessor;
+    private readonly int totalConsumerWorkers;
+    private readonly int maxInstanceWorkers;
+
+    public ConsumerLagWorkerBalancer(
+        IClusterManager clusterManager,
+        IConsumerAccessor consumerAccessor,
+        int totalConsumerWorkers,
+        int maxInstanceWorkers)
+    {
+        this.clusterManager = clusterManager;
+        this.consumerAccessor = consumerAccessor;
+        this.totalConsumerWorkers = totalConsumerWorkers;
+        this.maxInstanceWorkers = maxInstanceWorkers;
+    }
+
+    public async Task<int> GetWorkersCountAsync(WorkersCountContext context)
+    {
+        var topicsMetadata = await this.GetTopicsMetadataAsync(context);
+
+        var lastOffsets = this.GetPartitionsLastOffset(context.ConsumerName, topicsMetadata);
+
+        var partitionsOffset = await this.clusterManager.GetConsumerGroupOffsetsAsync(
+            context.ConsumerGroupId,
+            context.AssignedTopicsPartitions.Select(t => t.Name));
+
+        var partitionsLag = CalculatePartitionsLag(lastOffsets, partitionsOffset);
+        var myLag = CalculateMyPartitionsLag(context, partitionsLag);
+
+        decimal totalConsumerLag = Math.Max(partitionsLag.Sum(p => p.Lag), 1);
+
+        var ratio = myLag / totalConsumerLag;
+
+        var workers = (int)Math.Round(this.totalConsumerWorkers * ratio);
+
+        if (workers > this.maxInstanceWorkers)
+        {
+            return this.maxInstanceWorkers;
+        }
+
+        return workers < 1 ? 1 : workers;
+    }
+
+    private static long CalculateMyPartitionsLag(
+        WorkersCountContext context,
+        IReadOnlyList<(string Topic, int Partition, long Lag)> partitionsLag)
+    {
+        return partitionsLag
+            .Where(
+                partitionLag => context.AssignedTopicsPartitions
+                    .Any(
+                        topic => topic.Name == partitionLag.Topic &&
+                                 topic.Partitions.Any(p => p == partitionLag.Partition)))
+            .Sum(partitionLag => partitionLag.Lag);
+    }
+
+    private static IReadOnlyList<(string Topic, int Partition, long Lag)> CalculatePartitionsLag(
+        IEnumerable<(string Topic, int Partition, long Offset)> lastOffsets,
+        IEnumerable<TopicPartitionOffset> currentPartitionsOffset)
+    {
+        return lastOffsets
+            .Select(
+                last =>
+                {
+                    var currentOffset = currentPartitionsOffset
+                        .Where(current => current.Topic == last.Topic && current.Partition == last.Partition)
+                        .Select(current => current.Offset)
+                        .FirstOrDefault(0);
+
+                    var lastOffset = Math.Max(0, last.Offset);
+                    currentOffset = Math.Max(0, currentOffset);
+
+                    return (last.Topic, last.Partition, lastOffset - currentOffset);
+                })
+            .ToList();
+    }
+
+    private IReadOnlyList<(string TopicName, int Partition, long Offset)> GetPartitionsLastOffset(
+        string consumerName,
+        IEnumerable<(string Name, TopicMetadata Metadata)> topicsMetadata)
+    {
+        var consumer = this.consumerAccessor[consumerName];
+
+        var offsets = new List<(string TopicName, int Partition, long Offset)>();
+
+        foreach (var topic in topicsMetadata)
+        {
+            foreach (var partition in topic.Metadata.Partitions)
+            {
+                offsets.Add(
+                    (
+                        topic.Name,
+                        partition.Id,
+                        consumer.QueryWatermarkOffsets(
+                                new TopicPartition(topic.Name, new Partition(partition.Id)),
+                                TimeSpan.FromSeconds(30))
+                            .High.Value));
+            }
+        }
+
+        return offsets;
+    }
+
+    private async Task<IReadOnlyList<(string Name, TopicMetadata Metadata)>> GetTopicsMetadataAsync(WorkersCountContext context)
+    {
+        var topicsMetadata = new List<(string Name, TopicMetadata Metadata)>(context.AssignedTopicsPartitions.Count);
+
+        foreach (var topic in context.AssignedTopicsPartitions)
+        {
+            topicsMetadata.Add((topic.Name, await this.clusterManager.GetTopicMetadataAsync(topic.Name)));
+        }
+
+        return topicsMetadata;
+    }
+}

samples/KafkaFlow.Sample.Dashboard/Program.cs

@@ -1,13 +1,44 @@
+using System;
+using KafkaFlow.Producers;
 using KafkaFlow.Sample.Dashboard;
 using Microsoft.AspNetCore.Hosting;
+using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Hosting;
 
-await CreateHostBuilder(args)
-    .Build()
-    .RunAsync();
+var host = CreateHostBuilder(args)
+    .Build();
 
+var producer = host.Services.GetRequiredService<IProducerAccessor>()["producer"];
+
+_ = host.RunAsync();
 
 static IHostBuilder CreateHostBuilder(string[] args) =>
     Host
         .CreateDefaultBuilder(args)
-        .ConfigureWebHostDefaults(webBuilder => { webBuilder.UseStartup<Startup>(); });
+        .ConfigureWebHostDefaults(
+            webBuilder =>
+            {
+                webBuilder
+                    .UseStartup<Startup>();
+            });
+
+while (true)
+{
+    var input = Console.ReadLine();
+
+    var splitted = input.Split(" ");
+
+    var count = int.Parse(splitted[0]);
+
+    int? partition = null;
+
+    if (splitted.Length > 1)
+    {
+        partition = int.Parse(splitted[1]);
+    }
+
+    for (int i = 0; i < count; i++)
+    {
+        _ = producer.ProduceAsync(Guid.NewGuid().ToString(), Array.Empty<byte>(), partition: partition);
+    }
+}

samples/KafkaFlow.Sample.Dashboard/Startup.cs

@@ -1,6 +1,10 @@
 namespace KafkaFlow.Sample.Dashboard;
 
+using System;
+using System.Threading.Tasks;
 using KafkaFlow.Admin.Dashboard;
+using KafkaFlow.Clusters;
+using KafkaFlow.Consumers;
 using Microsoft.AspNetCore.Builder;
 using Microsoft.AspNetCore.Hosting;
 using Microsoft.Extensions.DependencyInjection;
@@ -11,31 +15,45 @@ public class Startup
     // This method gets called by the runtime. Use this method to add services to the container.
     public void ConfigureServices(IServiceCollection services)
     {
-        services.AddKafkaFlowHostedService(
-            kafka => kafka
-                .UseConsoleLog()
-                .AddCluster(
-                    cluster =>
-                    {
-                        const string topicName = "topic-dashboard";
-                        cluster
-                            .WithBrokers(new[] { "localhost:9092" })
-                            .EnableAdminMessages("kafka-flow.admin", "kafka-flow.admin.group.id")
-                            .EnableTelemetry("kafka-flow.admin", "kafka-flow.telemetry.group.id")
-                            .CreateTopicIfNotExists(topicName, 3, 1)
-                            .AddConsumer(
-                                consumer =>
-                                {
-                                    consumer
-                                        .Topics(topicName)
-                                        .WithGroupId("groupid-dashboard")
-                                        .WithName("consumer-dashboard")
-                                        .WithBufferSize(100)
-                                        .WithWorkersCount(20)
-                                        .WithAutoOffsetReset(AutoOffsetReset.Latest);
-                                });
-                    })
-        );
+        services
+            .AddKafkaFlowHostedService(
+                kafka => kafka
+                    .UseConsoleLog()
+                    .AddCluster(
+                        cluster =>
+                        {
+                            const string topicName = "topic-dashboard";
+                            cluster
+                                .WithBrokers(new[] { "localhost:9092" })
+                                .EnableAdminMessages("kafka-flow.admin", "kafka-flow.admin.group.id")
+                                .EnableTelemetry("kafka-flow.admin", "kafka-flow.telemetry.group.id")
+                                .CreateTopicIfNotExists(topicName, 3, 1)
+                                .AddConsumer(
+                                    consumer =>
+                                    {
+                                        consumer
+                                            .Topics(topicName)
+                                            .WithGroupId("groupid-dashboard")
+                                            .WithName("consumer-dashboard")
+                                            .WithBufferSize(20)
+                                            .WithManualStoreOffsets()
+                                            .WithWorkersCount(
+                                                (context, resolver) =>
+                                                    new ConsumerLagWorkerBalancer(
+                                                            resolver.Resolve<IClusterManager>(),
+                                                            resolver.Resolve<IConsumerAccessor>(),
+                                                            2,
+                                                            1)
+                                                        .GetWorkersCountAsync(context),
+                                                TimeSpan.FromSeconds(60))
+                                            .WithAutoOffsetReset(AutoOffsetReset.Latest)
+                                            .AddMiddlewares(
+                                                m =>
+                                                    m.Add<DelayMiddleware>());
+                                    })
+                                .AddProducer("producer", producer => producer.DefaultTopic(topicName));
+                        })
+            );
 
         services
             .AddControllers();
@@ -49,4 +67,14 @@ public void Configure(IApplicationBuilder app, IWebHostEnvironment env, IHostApp
             .UseEndpoints(endpoints => { endpoints.MapControllers(); })
             .UseKafkaFlowDashboard();
     }
-}
+}
+
+public class DelayMiddleware : IMessageMiddleware
+{
+    public async Task Invoke(IMessageContext context, MiddlewareDelegate next)
+    {
+        Console.WriteLine("message consumed {0}", context.Message.Key.GetHashCode());
+        _ = Task.Delay(100).ContinueWith(_ => context.ConsumerContext.StoreOffset());
+        await Task.CompletedTask;
+    }
+}

src/KafkaFlow.Abstractions/Configuration/IConsumerConfigurationBuilder.cs

@@ -2,6 +2,7 @@ namespace KafkaFlow.Configuration
 {
     using System;
     using System.Collections.Generic;
+    using System.Threading.Tasks;
 
     /// <summary>
     /// Used to build the consumer configuration
@@ -91,6 +92,23 @@ public interface IConsumerConfigurationBuilder
         /// <returns></returns>
         IConsumerConfigurationBuilder WithWorkersCount(int workersCount);
 
+        /// <summary>
+        /// Configures a custom function to dynamically calculate the number of workers.
+        /// </summary>
+        /// <param name="calculator">A function that takes a WorkersCountContext object and returns a Task yielding the new workers count</param>
+        /// <param name="evaluationInterval">The interval that the calculator will be called</param>
+        /// <returns>The IConsumerConfigurationBuilder instance for method chaining</returns>
+        IConsumerConfigurationBuilder WithWorkersCount(
+            Func<WorkersCountContext, IDependencyResolver, Task<int>> calculator,
+            TimeSpan evaluationInterval);
+
+        /// <summary>
+        /// Configures a custom function to dynamically calculate the number of workers.
+        /// </summary>
+        /// <param name="calculator">A function that takes a WorkersCountContext object and returns a Task yielding the new workers count</param>
+        /// <returns>The IConsumerConfigurationBuilder instance for method chaining</returns>
+        IConsumerConfigurationBuilder WithWorkersCount(Func<WorkersCountContext, IDependencyResolver, Task<int>> calculator);
+
         /// <summary>
         /// Sets how many messages will be buffered for each worker
         /// </summary>

src/KafkaFlow.Abstractions/Configuration/WorkersCountContext.cs

@@ -0,0 +1,29 @@
+namespace KafkaFlow.Configuration
+{
+    using System.Collections.Generic;
+
+    /// <summary>
+    /// A metadata class with some context information help to calculate the number of workers
+    /// </summary>
+    public class WorkersCountContext
+    {
+        public WorkersCountContext(
+            string consumerName,
+            string consumerGroupId,
+            IReadOnlyCollection<TopicPartitions> assignedTopicsPartitions)
+        {
+            this.ConsumerName = consumerName;
+            this.ConsumerGroupId = consumerGroupId;
+            this.AssignedTopicsPartitions = assignedTopicsPartitions;
+        }
+
+        public string ConsumerName { get; }
+
+        public string ConsumerGroupId { get; }
+
+        /// <summary>
+        /// Gets the assigned partitions to the consumer
+        /// </summary>
+        public IReadOnlyCollection<TopicPartitions> AssignedTopicsPartitions { get; }
+    }
+}

src/KafkaFlow.Abstractions/IConsumerContext.cs

@@ -53,6 +53,21 @@ public interface IConsumerContext
         /// </summary>
         bool ShouldStoreOffset { get; set; }
 
+        /// <summary>
+        /// Gets an instance of IDependencyResolver which provides methods to resolve dependencies.
+        /// This instance is tied to the consumer scope, meaning it is capable of resolving dependencies
+        /// that are scoped to the lifecycle of a single consumer.
+        /// </summary>
+        IDependencyResolver ConsumerDependencyResolver { get; }
+
+        /// <summary>
+        /// Gets an instance of IDependencyResolver which provides methods to resolve dependencies.
+        /// This instance is tied to the worker scope, meaning it is capable of resolving dependencies
+        /// that are scoped to the lifecycle of a single worker.
+        /// </summary>
+        IDependencyResolver WorkerDependencyResolver { get; }
+
+
         /// <summary>
         /// Store the message offset when manual store option is used
         /// </summary>

src/KafkaFlow.Abstractions/IDateTimeProvider.cs

@@ -10,6 +10,9 @@ public interface IDateTimeProvider
         /// <inheritdoc cref="DateTime.Now"/>
         DateTime Now { get; }
 
+        /// <inheritdoc cref="DateTime.UtcNow"/>
+        DateTime UtcNow { get; }
+
         /// <inheritdoc cref="DateTime.MinValue"/>
         DateTime MinValue { get; }
     }

src/KafkaFlow.Abstractions/IMessageContext.cs

@@ -27,6 +27,14 @@ public interface IMessageContext
         /// </summary>
         IProducerContext ProducerContext { get; }
 
+        /// <summary>
+        /// Gets an instance of IDependencyResolver which provides methods to resolve dependencies.
+        /// This instance is tied to the message scope, meaning it is capable of resolving dependencies
+        /// that are scoped to the lifecycle of a single processed message.
+        /// </summary>
+        IDependencyResolver DependencyResolver { get; }
+
+
         /// <summary>
         /// Creates a new <see cref="IMessageContext"/> with the new message
         /// </summary>