Dekaf - A .NET Kafka Client

Taking the Java out of Kafka.

Dekaf is a high-performance, pure C# Apache Kafka client for .NET 10+. No JVM, no interop, no native dependencies - just clean, modern C# all the way down.

If you like, or use this library, a sponsor is appreciated!

Benchmarks vs Confluent

View Full Documentation

Why Dekaf?

Unlike libraries that wrap librdkafka, Dekaf is a native .NET implementation with no delegation to other runtimes and unmanaged code:

• Pure C# - No native dependencies, no interop overhead
• Zero-allocation hot paths - Uses Span<T>, ref struct, and object pooling for minimal GC pressure
• Modern .NET - Built for .NET 10+ with nullable reference types, IAsyncEnumerable, and all the good stuff
• Native AOT compatible - Trim-safe and Native AOT ready, verified by CI smoke tests on every build
• Simple API - Intuitive fluent builders that do what you'd expect
• Batteries included for services - Hosted consumer base class with graceful shutdown, retries, retry topics, and dead letter queues built in

Getting Started

dotnet add package Dekaf

Producing Messages

The simplest way to send a message:

using Dekaf;

await using var producer = await Kafka.CreateProducer<string, string>()
    .WithBootstrapServers("localhost:9092")
    .BuildAsync();

// Wait for acknowledgment
var metadata = await producer.ProduceAsync("my-topic", "key", "Hello, Kafka!");
Console.WriteLine($"Sent to partition {metadata.Partition} at offset {metadata.Offset}");

For high-throughput scenarios where you don't need to wait for broker acknowledgment:

// Fire-and-forget - waits only for local enqueue/backpressure
await producer.FireAsync("my-topic", "key", "value");

// Make sure everything's delivered before shutting down
await producer.FlushAsync();

Topic-Specific Producers

When you're always producing to the same topic, use a topic producer for a cleaner API:

await using var producer = await Kafka.CreateProducer<string, string>()
    .WithBootstrapServers("localhost:9092")
    .BuildForTopicAsync("orders");

// No topic parameter needed
await producer.ProduceAsync("order-123", orderJson);
await producer.FireAsync("order-456", orderJson);

You can also create multiple topic producers that share the same connection:

await using var baseProducer = Kafka.CreateProducer<string, string>()
    .WithBootstrapServers("localhost:9092")
    .Build();

var orders = baseProducer.ForTopic("orders");
var events = baseProducer.ForTopic("events");

await orders.ProduceAsync("order-1", orderJson);
await events.ProduceAsync("event-1", eventJson);

Consuming Messages

using Dekaf;

await using var consumer = await Kafka.CreateConsumer<string, string>()
    .WithBootstrapServers("localhost:9092")
    .WithGroupId("my-consumer-group")
    .SubscribeTo("my-topic")
    .BuildAsync();

await foreach (var message in consumer.ConsumeAsync(cancellationToken))
{
    Console.WriteLine($"Got: {message.Key} = {message.Value}");
}

Configuration Presets

Not sure which settings to use? We've got you covered with presets for common scenarios:

using Dekaf;

// Maximize throughput (batching, compression, relaxed durability)
var producer = await Kafka.CreateProducer<string, string>()
    .WithBootstrapServers("localhost:9092")
    .ForHighThroughput()
    .BuildAsync();

// Minimize latency (no batching delay, smaller batches)
var producer = await Kafka.CreateProducer<string, string>()
    .WithBootstrapServers("localhost:9092")
    .ForLowLatency()
    .BuildAsync();

// Maximum reliability (all replicas must ack, idempotent)
var producer = await Kafka.CreateProducer<string, string>()
    .WithBootstrapServers("localhost:9092")
    .ForReliability()
    .BuildAsync();

You can override individual settings after applying a preset.

Batch Production

Need to send a bunch of messages? ProduceAllAsync handles the tricky ValueTask semantics for you:

var messages = new[]
{
    new ProducerMessage<string, string> { Topic = "orders", Key = "order-1", Value = orderJson1 },
    new ProducerMessage<string, string> { Topic = "orders", Key = "order-2", Value = orderJson2 },
    new ProducerMessage<string, string> { Topic = "orders", Key = "order-3", Value = orderJson3 },
};

var results = await producer.ProduceAllAsync(messages);

Or if all your messages go to the same topic:

var results = await producer.ProduceAllAsync("orders", new[]
{
    ("order-1", orderJson1),
    ("order-2", orderJson2),
    ("order-3", orderJson3),
});

Working with Headers

Headers are great for metadata like correlation IDs, trace context, or routing hints:

using Dekaf;

var headers = Headers.Create()
    .Add("correlation-id", correlationId)
    .Add("source", "order-service")
    .AddIfNotNull("user-id", userId)           // Only adds if not null
    .AddIf(isRetry, "retry-count", "1");       // Only adds if condition is true

await producer.ProduceAsync("orders", orderId, orderJson, headers);

Consumer LINQ Extensions

Process consumed messages with familiar LINQ-style operations:

// Filter and limit
await foreach (var message in consumer.ConsumeAsync(ct)
    .Where(m => m.Value.Contains("important"))
    .Take(100))
{
    await ProcessAsync(message);
}

// Batch processing - great for bulk database inserts
await foreach (var batch in consumer.ConsumeAsync(ct).Batch(100))
{
    await BulkInsertAsync(batch);
    await consumer.CommitAsync();
}

// Simple processing loop
await consumer.ForEachAsync(async msg =>
{
    await HandleMessageAsync(msg);
}, cancellationToken);

Note: filtered messages are skipped permanently once offsets are committed past them — see the docs before filtering messages you might need later!

Offset Management

Dekaf gives you control over when offsets are committed:

using Dekaf;

// Auto mode (default): Offsets committed automatically in the background
// Good for: Log processing, analytics, cases where losing a message is OK
var consumer = await Kafka.CreateConsumer<string, string>()
    .WithBootstrapServers("localhost:9092")
    .WithGroupId("my-group")
    .WithOffsetCommitMode(OffsetCommitMode.Auto)
    .BuildAsync();

// Manual mode: You control when to commit by calling CommitAsync()
// Dekaf tracks consumed offsets for you - CommitAsync() commits the latest
// consumed position for each partition. This gives you at-least-once semantics:
// if your app crashes before committing, messages will be redelivered on restart.
// Good for: Payment processing, order handling, anything where you can't lose messages
var consumer = await Kafka.CreateConsumer<string, string>()
    .WithBootstrapServers("localhost:9092")
    .WithGroupId("my-group")
    .WithOffsetCommitMode(OffsetCommitMode.Manual)
    .BuildAsync();

await foreach (var msg in consumer.ConsumeAsync(ct))
{
    await ProcessAsync(msg);
    await consumer.CommitAsync();  // Commits offset for all consumed messages
}

Compression

Dekaf supports standard Kafka compression codecs. Just add the relevant package:

dotnet add package Dekaf.Compression.Lz4     # Fast, good compression
dotnet add package Dekaf.Compression.Zstd    # Best compression ratio
dotnet add package Dekaf.Compression.Snappy  # Balanced

Then enable it:

using Dekaf;

var producer = await Kafka.CreateProducer<string, string>()
    .WithBootstrapServers("localhost:9092")
    .UseLz4Compression()
    .BuildAsync();

Serialization

Built-in serializers handle common types automatically:

• string
• byte[] and ReadOnlyMemory<byte>
• int, long, Guid

For JSON, add the serialization package:

dotnet add package Dekaf.Serialization.Json

using Dekaf;

var producer = await Kafka.CreateProducer<string, Order>()
    .WithBootstrapServers("localhost:9092")
    .WithValueSerializer(new JsonSerializer<Order>())
    .BuildAsync();

await producer.ProduceAsync("orders", order.Id, order);

Security

TLS

using Dekaf;

var producer = await Kafka.CreateProducer<string, string>()
    .WithBootstrapServers("kafka.example.com:9093")
    .UseTls()
    .BuildAsync();

SASL Authentication

using Dekaf;

// SASL/PLAIN
var producer = await Kafka.CreateProducer<string, string>()
    .WithBootstrapServers("kafka.example.com:9093")
    .UseTls()
    .WithSaslPlain("username", "password")
    .BuildAsync();

// SASL/SCRAM-SHA-512
var producer = await Kafka.CreateProducer<string, string>()
    .WithBootstrapServers("kafka.example.com:9093")
    .UseTls()
    .WithSaslScramSha512("username", "password")
    .BuildAsync();

Dependency Injection

For ASP.NET Core or other DI scenarios:

dotnet add package Dekaf.Extensions.DependencyInjection

using Dekaf.Extensions.DependencyInjection;

services.AddDekaf(dekaf =>
{
    dekaf.AddProducer<string, string>(producer => producer
        .WithBootstrapServers(configuration["Kafka:BootstrapServers"]!)
        .WithLinger(TimeSpan.FromMilliseconds(5))
        .ForReliability());

    dekaf.AddConsumer<string, string>(consumer => consumer
        .WithBootstrapServers(configuration["Kafka:BootstrapServers"]!)
        .WithGroupId("my-service")
        .WithFetchMinBytes(1024));
});

Then inject IKafkaProducer<string, string> or IKafkaConsumer<string, string> wherever you need them. The DI callbacks use the same ProducerBuilder and ConsumerBuilder types as Kafka.CreateProducer() and Kafka.CreateConsumer(), so advanced options, SASL/TLS, retry policies, presets, interceptors, and compression extension methods are available there too.

Global Interceptors

Register cross-cutting interceptors (tracing, metrics, audit logging) that apply to all producers or consumers:

services.AddDekaf(dekaf =>
{
    // Global interceptors apply to every producer/consumer
    dekaf.AddGlobalProducerInterceptor(typeof(TracingInterceptor<,>));
    dekaf.AddGlobalConsumerInterceptor(typeof(MetricsInterceptor<,>));

    dekaf.AddProducer<string, string>(producer => producer
        .WithBootstrapServers("localhost:9092")
        .WithLinger(TimeSpan.FromMilliseconds(5)));

    dekaf.AddConsumer<string, string>(consumer => consumer
        .WithBootstrapServers("localhost:9092")
        .WithGroupId("my-service")
        .WithPrefetchPipelineDepth(4));
});

Global interceptors execute before per-instance interceptors, in registration order. They are constructed via ActivatorUtilities, so their dependencies are resolved from the DI container.

Hosted Consumer Services

For background consumers in ASP.NET Core or Worker Service apps, skip the hand-rolled consume loop. Dekaf.Extensions.Hosting provides KafkaConsumerService<TKey, TValue>, which handles the loop, graceful shutdown (drain + final commit), in-place retries, and optional tiered retry topics and dead letter queue routing — you just override ProcessAsync:

dotnet add package Dekaf.Extensions.Hosting

public sealed class OrderProcessor : KafkaConsumerService<string, Order>
{
    public OrderProcessor(
        IKafkaConsumer<string, Order> consumer,
        ILogger<OrderProcessor> logger,
        DeadLetterOptions? deadLetterOptions = null)   // forwarded so the DLQ callback below takes effect
        : base(consumer, logger, deadLetterOptions)
    {
    }

    protected override IEnumerable<string> Topics => ["orders"];

    protected override async ValueTask ProcessAsync(
        ConsumeResult<string, Order> result, CancellationToken cancellationToken)
    {
        await HandleOrderAsync(result.Value, cancellationToken);
    }
}

services.AddDekaf(dekaf =>
{
    dekaf.AddConsumerService<OrderProcessor, string, Order>(
        consumer => consumer
            .WithBootstrapServers("localhost:9092")
            .WithGroupId("orders-service"),
        dlq => dlq.WithRetryTopics(TimeSpan.FromSeconds(5), TimeSpan.FromSeconds(30)));  // optional DLQ + retry tiers
});

Failed messages flow orders → orders-retry-5s → orders-retry-30s → orders.DLQ, carrying headers with the source offset and failure details. See the Hosted Consumer Services and Dead Letter Queues docs.