No history yet

Kafka Client Infrastructure

Connecting .NET to Kafka

To get started, you'll need the Confluent.Kafka NuGet package. This is the standard, high-performance client library for using Kafka in a .NET application. It's built on top of librdkafka, a C++ library, which gives it a significant performance edge.

dotnet add package Confluent.Kafka

Once the package is installed, the first step is configuring the connection. All client configurations, whether for a producer or a consumer, start with the ClientConfig class. This object holds the common settings needed to talk to your Kafka cluster.

Three properties are fundamental:

PropertyDescription
BootstrapServersA comma-separated list of broker addresses (host:port). The client only needs one to connect, as it will discover the rest of the cluster automatically.
SecurityProtocolSpecifies the protocol used to communicate with brokers. Common values are Plaintext, Ssl, SaslPlaintext, or SaslSsl.
SaslMechanismDefines the authentication method if you're using SASL. Examples include Plain, ScramSha256, or ScramSha512.

For a production environment, you'll almost always connect securely. Here’s how you would set up a ClientConfig for a secure connection using SASL over SSL.

using Confluent.Kafka;

var config = new ClientConfig
{
    BootstrapServers = "your-kafka-broker-1:9092,your-kafka-broker-2:9092",
    SecurityProtocol = SecurityProtocol.SaslSsl,
    SaslMechanism = SaslMechanism.Plain,
    SaslUsername = "your-username",
    SaslPassword = "your-password"
};

Building Producers and Consumers

With a base configuration in place, you can create producers and consumers. These have their own specific configuration classes, ProducerConfig and ConsumerConfig, which inherit from ClientConfig. This lets you add producer or consumer-specific settings.

A producer is constructed using ProducerBuilder. It takes the ProducerConfig and builds an IProducer instance. This client is thread-safe and designed to be long-lived.

using Confluent.Kafka;

// Assumes 'config' is a ClientConfig or ProducerConfig object
var producerConfig = new ProducerConfig(config);

var producer = new ProducerBuilder<string, string>(producerConfig).Build();

Similarly, a consumer is built with ConsumerBuilder. A key difference is that you must provide a GroupId. This identifier tells Kafka which consumer group this instance belongs to, which is crucial for coordinating how topic partitions are distributed among consumers.

using Confluent.Kafka;

var consumerConfig = new ConsumerConfig(config)
{
    GroupId = "my-processing-group",
    AutoOffsetReset = AutoOffsetReset.Earliest
};

var consumer = new ConsumerBuilder<string, string>(consumerConfig).Build();

The methods for sending messages (ProduceAsync) and receiving them (Consume) are asynchronous. They return Task objects, fitting naturally into .NET's Task Parallel Library (TPL) for non-blocking I/O operations.

Dependency Injection Patterns

In a modern .NET application, you shouldn't be creating clients manually with new. Instead, you should use the built-in dependency injection (DI) container to manage their lifetimes. The producer and consumer have different lifecycle requirements.

The IProducer instance is thread-safe and efficient. The underlying connection and resources are managed for you. Because creating it is relatively expensive, it's a perfect candidate for a singleton. You register it once and inject it wherever you need to send messages.

By decoupling data producers from consumers, Kafka allows scalable collection and transformation of massive data streams, ensuring that information flows from edge devices to analytic platforms without delay.

The IConsumer is a different story. It is not thread-safe. You cannot share a single consumer instance across multiple concurrent threads, like you might in an ASP.NET Core web request. The standard practice is to create a consumer that lives inside a long-running background service, often an IHostedService.

This background service runs a continuous loop that calls consumer.Consume(), processes the message, and then repeats. This isolates the consumer to a single, dedicated thread, avoiding concurrency issues.

Here’s how you might register these components in your Program.cs:

// In Program.cs or Startup.cs

using Confluent.Kafka;

var builder = WebApplication.CreateBuilder(args);

// 1. Configure and register the IProducer as a singleton
var producerConfig = new ProducerConfig();
builder.Configuration.GetSection("Kafka:Producer").Bind(producerConfig);

builder.Services.AddSingleton<IProducer<string, string>>(
    _ => new ProducerBuilder<string, string>(producerConfig).Build()
);

// 2. Register the consumer-hosting background service
builder.Services.AddHostedService<MyKafkaConsumerService>();

// In a separate file: MyKafkaConsumerService.cs
public class MyKafkaConsumerService : BackgroundService
{
    private readonly IConsumer<string, string> _consumer;

    public MyKafkaConsumerService(IConfiguration configuration)
    {
        var consumerConfig = new ConsumerConfig();
        configuration.GetSection("Kafka:Consumer").Bind(consumerConfig);
        _consumer = new ConsumerBuilder<string, string>(consumerConfig).Build();
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        _consumer.Subscribe("my-topic");

        while (!stoppingToken.IsCancellationRequested)
        {
            var consumeResult = _consumer.Consume(stoppingToken);
            // Process the message...
        }

        _consumer.Close();
    }
}

This approach correctly manages the client lifecycles, ensuring your application is both performant and stable.