Open.ChannelExtensions.Kafka 1.0.2

Open.ChannelExtensions.Kafka

Bridges Confluent.Kafka with System.Threading.Channels and Open.ChannelExtensions, giving you resilient, back-pressure-aware Kafka producers and consumers with minimal ceremony.

Features

• Consumer → ChannelReader — pipe Kafka messages into a bounded channel with built-in back-pressure.
• Consumer → IAsyncEnumerable — iterate consumed messages with optional buffering. Unbuffered mode is LINQ-compatible.
• Resilient consumption — ConsumeUntilCancelled automatically reconnects and resubscribes after transient failures with built-in back-off delays.
• Managed producer singleton — thread-safe, lazily-built IProducer with automatic rebuild on fatal errors and exponential back-off.
• Dependency injection — register a Kafka producer as a singleton service in one call.
• Structured logging — optional ILogger integration throughout.
• Broad target support — netstandard2.0, netstandard2.1, net8.0, net9.0.

Installation

dotnet add package Open.ChannelExtensions.Kafka

Or via the Package Manager Console:

Install-Package Open.ChannelExtensions.Kafka

Quick Start

Consuming

As a ChannelReader

Turn any IConsumer<TKey, TValue> into a bounded ChannelReader that fills in the background:

using Confluent.Kafka;
using Open.ChannelExtensions.Kafka;

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

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

// Bounded channel with capacity of 100 messages.
ChannelReader<ConsumeResult<string, string>> reader =
    consumer.ToChannelReader(capacity: 100, cancellationToken: cts.Token);

// Now use Open.ChannelExtensions to process:
await reader
    .ReadAllAsync(cts.Token)
    .ForEachAsync(result =>
    {
        Console.WriteLine($"{result.Message.Key}: {result.Message.Value}");
    });

The singleReader parameter (default false) can be set to true when only one thread will read from the channel — this enables internal optimizations in the channel implementation.

As an IAsyncEnumerable

Buffered — reads ahead into an internal channel for throughput:

await foreach (var result in consumer.ToBufferedAsyncEnumerable(bufferSize: 50, cancellationToken: cts.Token))
{
    Console.WriteLine(result.Message.Value);
}

Note: Because the buffer continues to fill independently, this is not LINQ-compatible.

Unbuffered — yields one at a time. Safe for LINQ composition:

await foreach (var result in consumer.AsAsyncEnumerable(logger: null, cancellationToken: cts.Token))
{
    Console.WriteLine(result.Message.Value);
}

Resilient consumption with ConsumeUntilCancelled

For long-lived consumers that must survive broker restarts, network blips, and transient errors:

Func<ConsumerBuilder<string, string>> builderFactory =
    () => new ConsumerBuilder<string, string>(config);

await foreach (var result in builderFactory.ConsumeUntilCancelled("my-topic", logger, cts.Token))
{
    Process(result);
}

This method:

• Rebuilds the consumer if construction or subscription fails (retries after 30 seconds).
• Restarts after a consume error (retries after 5 seconds).
• Stops cleanly when the CancellationToken is cancelled.

You can also pass multiple topics:

builderFactory.ConsumeUntilCancelled(["topic-a", "topic-b"], logger, cts.Token);

Producing

Registering via Dependency Injection

Register a managed producer singleton in your DI container:

// Using a ProducerConfig:
services.AddKafkaProducer<string, string>(sp =>
    new ProducerConfig { BootstrapServers = "localhost:9092" });

// Or using a ProducerBuilder factory for full control:
services.AddKafkaProducer<string, string>(sp =>
    () => new ProducerBuilder<string, string>(
        new ProducerConfig { BootstrapServers = "localhost:9092" }));

Then inject IKafkaProducerProvider<string, string> and get the producer:

public class MyService(IKafkaProducerProvider<string, string> producerProvider)
{
    public async Task PublishAsync(string key, string value, CancellationToken ct)
    {
        var producer = await producerProvider.GetProducerAsync();
        await producer.ProduceAsync("my-topic", new Message<string, string>
        {
            Key = key,
            Value = value
        }, ct);
    }
}

Using KafkaProducerSingleton directly

If you're not using DI, you can create the singleton yourself:

using var producerSingleton = new KafkaProducerSingleton<string, string>(
    new ProducerConfig { BootstrapServers = "localhost:9092" },
    logger: null);

var producer = await producerSingleton.GetProducerAsync();
await producer.ProduceAsync("my-topic", new Message<string, string>
{
    Key = "key",
    Value = "hello"
});

The singleton:

• Lazily builds the IProducer on first access.
• Automatically disposes and rebuilds the producer when a fatal Kafka error is reported.
• Applies exponential back-off (5 → 10 → 20 → 40 seconds) between rebuild attempts, resetting after 5 minutes of stability.
• Is fully thread-safe — concurrent callers share the same producer instance.

API Reference

KafkaConsumerExtensions

KafkaConfigExtensions

KafkaProducerSingleton<TKey, TValue>

Requirements

• .NET Standard 2.0+, .NET 8.0+, or .NET 9.0+
• Confluent.Kafka ≥ 2.13.2
• Open.ChannelExtensions ≥ 9.3.0

License

MIT