[Bug Report]: ProducerConfigurationBuilder.Build does not respect ProducerConfig linger.ms and statistics.interval.ms

[mbaturin-rubbi]

Prerequisites

• I have searched issues to ensure it has not already been reported

Description

The linger.ms and statistics.interval.ms values specified in ProducerConfig are overwritten with empty values if the WithLingerMs and WithStatisticsIntervalMs methods are not called.

kafkaflow/src/KafkaFlow/Configuration/ProducerConfigurationBuilder.cs

Line 89 in 2596598

_producerConfig.StatisticsIntervalMs = _statisticsInterval;

Steps to reproduce

// PackageReference Include="KafkaFlow.Microsoft.DependencyInjection" Version="4.1.0"

using System.Reflection;
using Confluent.Kafka;
using KafkaFlow;
using KafkaFlow.Configuration;
using KafkaFlow.Producers;

var serviceCollection = new ServiceCollection();

var parameters = new Dictionary<string, string>
{
    { "linger.ms", "123" },
    { "statistics.interval.ms", "321" }
};

var producerConfig = new ProducerConfig(parameters);

Console.WriteLine($"LingerMs is {producerConfig.LingerMs}"); // LingerMs is 123
Console.WriteLine($"StatisticsIntervalMs is {producerConfig.StatisticsIntervalMs}"); // LingerMs is 321

serviceCollection.AddKafka(kafka =>
{
    kafka.AddCluster(cluster =>
        cluster
            .WithBrokers(["localhost:9092"])
            .AddProducer(
                "producer",
                configurationBuilder => configurationBuilder.WithProducerConfig(producerConfig)
            )
    );
});

var serviceProvider = serviceCollection.BuildServiceProvider();

var producer = serviceProvider.GetRequiredService<IProducerAccessor>().GetProducer("producer");

// ;)

var messageProducerType = producer.GetType();
var producerConfigurationField = messageProducerType.GetField(
    "_configuration",
    BindingFlags.Instance | BindingFlags.NonPublic
);

var newProducerConfig = (
    (IProducerConfiguration)producerConfigurationField.GetValue(producer)
).BaseProducerConfig;

Console.WriteLine($"Actual LingerMs is {newProducerConfig.LingerMs?.ToString() ?? "null"}"); // LingerMs is null
Console.WriteLine($"Actual StatisticsIntervalMs is {newProducerConfig.StatisticsIntervalMs}"); // StatisticsIntervalMs is 0

Expected behavior

If WithLingerMs and WithStatisticsIntervalMs are not called, the original values from ProducerConfig.LingerMs and ProducerConfig.StatisticsIntervalMs should be used.

Actual behavior

linger.ms is null, statistics.interval.ms is 0.

KafkaFlow version

4.1.0 (since v1.4.8)