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

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

Apache Kafka is a distributed streaming platform widely used for building high-load messaging systems, event processing, and microservice integration. For working with Kafka in the .NET ecosystem, the primary and most supported library is Confluent.Kafka — the official client from Confluent, built on top of the high-performance C library librdkafka.

In this article, we will explore the key features of Confluent.Kafka, installation, producer and consumer configuration, error handling, and provide practical code examples in C#.

Installation and Setup

The library is distributed via NuGet. To install, run the command:

dotnet add package Confluent.Kafka

After installation, add the using directives to your code:

using Confluent.Kafka;using Confluent.Kafka.Admin;using System;using System.Threading;using System.Threading.Tasks;

Basic Concepts

• Producer — sends messages to Kafka topics.
• Consumer — reads messages from topics.
• Topic — a named channel where messages are published.
• Broker — a Kafka server that stores data.
• Partition — a unit of parallelism within a topic.

Creating a Producer

The producer sends messages to Kafka. Consider a basic example:

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

using (var producer = new ProducerBuilder<string, string>(config).Build()){    try    {        var deliveryResult = await producer.ProduceAsync(            "my-topic",            new Message<string, string> { Key = "key1", Value = "Hello Kafka from .NET!" }        );

        Console.WriteLine($"Sent to partition [{deliveryResult.Partition}] at offset [{deliveryResult.Offset}]");    }    catch (ProduceException<string, string> e)    {        Console.WriteLine($"Send error: {e.Error.Reason}");    }}

In this example:

• BootstrapServers — the address of the Kafka cluster.
• ProduceAsync — asynchronous message sending.
• Key — optional key for partition determination.
• Value — the message body (can be JSON, string, binary data).

Creating a Consumer

The consumer subscribes to topics and processes messages. An important parameter is GroupId, which defines the consumer group:

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

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 message: Key = {consumeResult.Message.Key}, Value = {consumeResult.Message.Value}");

            // Manual commit            consumer.Commit(consumeResult);        }    }    catch (OperationCanceledException)    {        Console.WriteLine("Consumption stopped.");    }    finally    {        consumer.Close();    }}

Key points:

• AutoOff