FluentKafkaProducer 1.0.0

FluentKafkaProducer

This package provides a robust, extensible Kafka producer client for .NET applications. It includes serialization support for multiple formats (e.g., JSON, MessagePack), observability via OpenTelemetry metrics and activities, and a clean DI-friendly architecture.

✨ Features

• ✅ Plug-and-play integration with Kafka using Confluent's client

• ✅ Built-in support for JSON and MessagePack serialization

• ✅ Easily extendable with custom serializers

• ✅ OpenTelemetry-based metrics and tracing instrumentation

📦 Installation

Install

Install via NuGet:

dotnet add package FluentKafkaProducer

Or via the NuGet Package Manager:

Install-Package FluentKafkaProducer

🛠️ Usage

Register the Producer

    services.AddFluentKafkaProducerServices(new KafkaProducerSettingsDto
    {
        Hosts = "localhost:9092",
        UserName = "your-user",
        Password = "your-password",
        SaslMechanism = SaslMechanism.Plain,
        SecurityProtocol = SecurityProtocol.SaslPlaintext
    });

Producing messages

Inject IKafkaProducerService and send messages easily:

public class SomeService
{
    private readonly IKafkaProducerService _kafkaProducer;

    public SomeService(IKafkaProducerService kafkaProducer)
    {
        _kafkaProducer = kafkaProducer;
    }

    public async Task PublishAsync()
    {
        var message = new MyEvent { Id = 1, Name = "test" };
        await _kafkaProducer.SendToKafka(message, "my-topic", key: "event-1", formatKey: KafkaFormatKeys.Json);
    }
}

⚙️ KafkaProducerSettings

public sealed record KafkaProducerSettings
{
    public string Hosts { get; set; }
    public string UserName { get; set; }
    public string Password { get; set; }
    public SecurityProtocol SecurityProtocol { get; set; } = SecurityProtocol.SaslPlaintext;
    public SaslMechanism SaslMechanism { get; set; } = SaslMechanism.Plain;
}

Serialization

Two built-in serializers are included:

JsonKafkaMessageSerializer

MessagePackKafkaMessageSerializer

You can add your own by implementing:

public interface IKafkaMessageSerializer
{
    string FormatKey { get; }
    Headers Headers { get; }
    byte[] Serialize<T>(T data) where T : class;
}

Then register it:

services.AddSingleton<IKafkaMessageSerializer, CustomKafkaMessageSerializer>();

Use it when producing:

await _kafkaProducer.SendToKafka(message, "my-topic", key: "event-1", formatKey: "CustomFormatKey");

📊 Observability

OpenTelemetry instrumentation includes:

• ActivitySource: Tracing message send operations

• Metrics:

  • kafka.messages.sent

  • kafka.messages.failed

  • kafka.message.send.duration (ms)

Tracing includes:

• Kafka topic

• Message key

• Exceptions and status

builder.Services.AddOpenTelemetry()
    .WithTracing(b => b.AddKafkaProducerInstrumentation())
    .WithMetrics(b => b.AddKafkaProducerInstrumentation());

📌 Notes

• Message serialization is format-key based, defaulting to JSON.

• serializers are resolved by key; you can plug in any format.

• OpenTelemetry integration is optional but strongly recommended.

📜 License

MIT — Feel free to use and modify.