KafkaStorm

Simple .NET client for Kafka based on Confluent.Kafka.

Table of contents

• Requirements
• Features
• Installation
• Setup
• Assembly scanning
• Consuming
• Producing
• Examples
• Development
• Migrating from 9.x
• Related

Requirements

• .NET 10 SDK (LTS)

Features

• In-memory queue for messages that couldn't be sent
• Concurrent consumers via hosted services
• Producing messages concurrently
• JSON serialization via System.Text.Json (no Schema Registry required)

Installation

Package Manager

Install-Package KafkaStorm -Version 10.0.0

.NET CLI

dotnet add package KafkaStorm --version 10.0.0

Setup

using Confluent.Kafka;
using KafkaStorm.Extensions;
using KafkaStorm.Interfaces;

builder.Services.AddKafkaStorm(factory =>
{
    factory.AddProducer(prf =>
    {
        prf.ConfigProducer(new ProducerConfig
        {
            BootstrapServers = host
        });

        prf.InMemoryQueue();
        prf.SetQueueLimit(65536);
    });

    // Start the background worker that drains the retry queue:
    factory.StartProducerHostedService();

    factory.AddConsumers(crf =>
    {
        crf.AddConsumer<HelloConsumer, HelloEvent>(new ConsumerConfig
        {
            BootstrapServers = "localhost:29092",
            GroupId = "TestGroup"
        }, "topicName");
    });
});

Uses the same ConsumerConfig and ProducerConfig types as Confluent.Kafka.

Assembly scanning

Register all IConsumer<T> / IMessage pairs from an assembly:

using System.Reflection;
using Confluent.Kafka;
using KafkaStorm.Extensions;
using KafkaStorm.Interfaces;

builder.Services.AddKafkaStorm(factory =>
{
    factory.AddConsumers(crf =>
    {
        var config = new ConsumerConfig
        {
            BootstrapServers = "localhost:29092",
            GroupId = "TestGroup"
        };

        crf.AddConsumersFromAssembly(Assembly.GetExecutingAssembly(), config);
    });
});

Message types used with assembly scanning must implement IMessage.

Consuming

using KafkaStorm.Interfaces;
using Microsoft.Extensions.Logging;

public class HelloConsumer(ILogger<HelloConsumer> logger) : IConsumer<HelloEvent>
{
    public async Task Handle(HelloEvent @event, CancellationToken cancellationToken)
    {
        logger.LogDebug("Message received");
        await Task.CompletedTask;
    }
}

Event (message)

public class HelloEvent(DateTime time)
{
    public string Message { get; } = "Hello";
    public DateTime Time { get; } = time;
}

Note: Properties typed as interfaces may cause JSON deserialization errors.

Producing

Inject IProducer:

using KafkaStorm.Interfaces;

public class TestService(IProducer producer)
{
    // Queue on failure (when in-memory queue is enabled):
    public Task QueueProduce() => producer.Produce(new HelloEvent(DateTime.Now), "topicName");

    // Send immediately:
    public Task SendNow() => producer.ProduceNowAsync(new HelloEvent(DateTime.Now), "topicName");
}

Examples

See samples/BasicWebApi for a minimal Web API that produces and consumes Kafka messages with KafkaStorm.

Development

git clone https://github.com/stormaref/KafkaStorm.git
cd KafkaStorm
dotnet restore
dotnet build -c Release
dotnet test -c Release

See CONTRIBUTING.md for contribution guidelines.

Migrating from 9.x

• Target .NET 10 and use package version 10.0.0.
• Public registration APIs (AddKafkaStorm, AddProducer, AddConsumer, etc.) are unchanged.
• Breaking: Direct use of static fields on ProducerRegistrationFactory or ConsumerRegistrationFactory is no longer supported. Use normal DI registration instead.

See CHANGELOG.md for full release notes.

Author

@stormaref

Related

• Confluent's .NET Client