Dekaf - A .NET Kafka Client. Taking the Java out of Kafka.

Kafka in pure C#, without librdkafka.

Dekaf is a high-performance, pure C# Apache Kafka client optimized for .NET 10, with compatible package assets for earlier runtimes. Unlike Confluent.Kafka, which wraps the native librdkafka library, Dekaf implements the Kafka wire protocol in C#: no librdkafka, no native interop, no native dependencies.

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 - Optimized for .NET 10 with nullable reference types and IAsyncEnumerable; see package compatibility below
• Native AOT coverage - .NET 10 linux-x64 smoke and Kafka integration tests in code-change CI; see coverage below
• 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

Package and runtime compatibility

.NET 10 is the preferred runtime for Dekaf's performance-oriented APIs. Package assets and tested runtimes are separate:

Packages	Target frameworks shipped in NuGet
Dekaf, Dekaf.Abstractions	net10.0, net8.0, netstandard2.0
Dekaf.Compression.*, Dekaf.Serialization.*, Dekaf.SchemaRegistry and its extensions, Dekaf.Extensions.*, Dekaf.OpenTelemetry, Dekaf.Outbox and its EF Core and DynamoDB extensions, Dekaf.Testing, Dekaf.Aspire, Dekaf.Aspire.SchemaRegistry, Dekaf.Aspire.Hosting	net8.0, net10.0

• Tested runtimes: CI runs unit tests on .NET 8 and .NET 10. PR Kafka integration tests run on .NET 10, with GSSAPI package round trips on both runtimes; the NuGet release gate tests both runtimes against Kafka 4.0.2, 4.1.2, 4.2.1, 4.3.1 and 4.4.0.
• Development: use the .NET 10 SDK to build this repository. Running its net8.0 tests also requires the .NET 8 runtime (or SDK).
• Native AOT: code-change CI publishes and runs core and dependency-injection smoke apps plus a Kafka integration smoke subset on .NET 10 linux-x64. Release validation expands AOT integration coverage against Kafka 4.0.2 and 4.4.0. This is coverage of those scenarios, not every package/API/platform.

A netstandard2.0 asset does not mean every compatible runtime is tested or has identical APIs and performance. .NET 8 uses the dedicated net8.0 asset; optional packages require their own compatible assets. See API and runtime compatibility for asset differences, package validation and AOT limits.

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;
using Dekaf.Compression;
using Dekaf.Compression.Lz4;

CompressionCodecRegistry.Default.AddLz4();

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

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

// Maximum reliability (all replicas must ack, idempotent)
var reliableProducer = 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 autoCommitConsumer = 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 manualCommitConsumer = await Kafka.CreateConsumer<string, string>()
    .WithBootstrapServers("localhost:9092")
    .WithGroupId("my-group")
    .WithOffsetCommitMode(OffsetCommitMode.Manual)
    .BuildAsync();

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

Compression

Dekaf supports standard Kafka compression codecs. Install 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

Register the codec at startup, then select the producer's compression type:

using Dekaf;
using Dekaf.Compression;
using Dekaf.Compression.Lz4;

CompressionCodecRegistry.Default.AddLz4();

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

Consumer applications must also register every optional codec they read before building clients, using CompressionCodecRegistry.Default.AddLz4(), AddZstd(), or AddSnappy(). The producer's ForHighThroughput() preset selects LZ4 and needs the same registration. Gzip is built in and needs no extra package or registration. See the compression guide for consumer examples and imports.

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);

For heterogeneous event streams, Dekaf.Serialization.Routing provides frozen, allocation-free topic and schema-ID deserializers plus topic and runtime-type serializers.

Security

TLS

using Dekaf;

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

SASL Authentication

using Dekaf;

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

// SASL/SCRAM-SHA-512
var scramSha512Producer = 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

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
});

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);
    }
}

Failed messages flow orders → orders-retry-5s → orders-retry-30s → orders.DLQ, carrying headers with the source offset and failure details. If no retry or durable routing path succeeds, the service keeps running and redelivers the source record with backoff by default (MessageFailureDisposition.Redeliver); override GetFailureDispositionAsync to explicitly discard selected failures or to stop the service with Retry. Hosted consumer services require after-processing offset staging and reject WithAtMostOnceProcessing(). See the Hosted Consumer Services and Dead Letter Queues docs.