Working with Apache Kafka in C# (.NET): Confluent.Kafka Library

Confluent.Kafka Library for Working with Apache Kafka in C# (.NET)

Apache Kafka is a distributed streaming platform that has become the standard for building event-driven architectures and microservices. For C# (.NET) developers, the most popular and supported library is Confluent.Kafka — an official client from Confluent, built on top of the high-performance C-client librdkafka. In this article, we will cover in detail how to install the library, create a producer and consumer, and discuss best practices.

Installing Confluent.Kafka

To get started, add the package to your project via NuGet:

dotnet add package Confluent.Kafka

Or via the Package Manager Console:

Install-Package Confluent.Kafka

Basic Producer Setup

The producer sends messages to Kafka topics. Here is a simple example:

using Confluent.Kafka;

var config = new ProducerConfig{    BootstrapServers = "localhost:9092"};

using (var producer = new ProducerBuilder<string, string>(config).Build()){    var message = new Message<string, string>    {        Key = "user_123",        Value = "Hello, Kafka!"    };

    var deliveryReport = await producer.ProduceAsync("my-topic", message);    Console.WriteLine($"Message sent. Partition: {deliveryReport.Partition}, Offset: {deliveryReport.Offset}");}

Key points:

• BootstrapServers — address of one or more Kafka brokers.
• ProduceAsync — asynchronous send with acknowledgment waiting.
• Key allows guaranteeing message ordering by the same key.

Producer with Complex Object Serialization

Often you need to send JSON or binary data instead of strings. Use a custom serializer or the Newtonsoft.Json library:

using System.Text.Json;using Confluent.Kafka;

var config = new ProducerConfig { BootstrapServers = "localhost:9092" };

var producer = new ProducerBuilder<string, Order>(config)    .SetValueSerializer(new JsonSerializer<Order>())    .Build();

var order = new Order { Id = 1, Amount = 150.50m, Customer = "Ivan" };await producer.ProduceAsync("orders", new Message<string, Order> { Key = order.Id.ToString(), Value = order });

public class Order{    public int Id { get; set; }    public decimal Amount { get; set; }    public string Customer { get; set; }}

public class JsonSerializer<T> : ISerializer<T>{    public byte[] Serialize(T data, SerializationContext context) =>        JsonSerializer.SerializeToUtf8Bytes(data);}

Consumer Setup

The consumer reads messages from one or more topics. Example of subscribing and processing:

using Confluent.Kafka;

var config = new ConsumerConfig{    BootstrapServers = "localhost:9092",    GroupId = "my-consumer-group",    AutoOffsetReset = AutoOffsetReset.Earliest};

using (var consumer = new ConsumerBuilder<string, string>(config).Build()){    consumer.Subscribe("my-topic");

    CancellationTokenSource cts = new CancellationTokenSource();    Console.CancelKeyPress += (_, e) => { e.Cancel = true; cts.Cancel(); };

    try    {        while (true)        {            var consumeResult = consumer.Consume(cts.Token);            Console.WriteLine($"Received: Key={consumeResult.Message.Key}, Value={consumeResult.Message.Value}");        }    }    catch (OperationCanceledException)    {        consumer.Close();    }}

Important configuration parameters:

• GroupId — consumer group identifier. Kafka distributes partitions among