Dekaf

https://thomhurst.github.io/Dekaf/docs
Admin
Dekaf is a high-performance, pure C# Apache Kafka client for modern .NET that features...
Tokens:78,538
Snippets:955
Trust Score:9.7
Update:2 months ago
Show doc for...
Tokens:
Raw
### Install Dekaf.Extensions.Hosting

Source: https://thomhurst.github.io/Dekaf/docs/hosted-services

Add the required NuGet package to your .NET project.

```bash
dotnet add package Dekaf.Extensions.Hosting
```

--------------------------------

### Install Dekaf OpenTelemetry package

Source: https://thomhurst.github.io/Dekaf/docs/observability

Use the .NET CLI to add the instrumentation package to your project.

```bash
dotnet add package Dekaf.OpenTelemetry
```

--------------------------------

### Install Routing Package

Source: https://thomhurst.github.io/Dekaf/docs/serialization/routing

Add the routing package to your .NET project.

```bash
dotnet add package Dekaf.Serialization.Routing
```

--------------------------------

### Install Compression Packages

Source: https://thomhurst.github.io/Dekaf/docs/compression

Use the .NET CLI to add the required NuGet package for the desired compression codec.

```bash
# Recommended for most use cases

dotnet add package Dekaf.Compression.Lz4



# Best compression ratio

dotnet add package Dekaf.Compression.Zstd



# Alternative fast codec

dotnet add package Dekaf.Compression.Snappy
```

--------------------------------

### Configure Producer with Compression and Batching

Source: https://thomhurst.github.io/Dekaf/docs/compression

Example of setting up a producer with LZ4 compression, linger time, and batch size.

```csharp
using Dekaf;



var producer = await Kafka.CreateProducer<string, string>()

    .WithBootstrapServers("localhost:9092")

    .UseLz4Compression()

    .WithLingerMs(5)        // Wait up to 5ms to fill batches

    .WithBatchSize(65536)   // 64KB batches

    .BuildAsync();
```

--------------------------------

### Key-Based Partitioning Example

Source: https://thomhurst.github.io/Dekaf/docs/producer/partitioning

Messages sharing the same key are routed to the same partition to ensure sequential processing.

```csharp
// All messages for order-123 go to the same partition

await producer.ProduceAsync("orders", "order-123", event1);

await producer.ProduceAsync("orders", "order-123", event2);

await producer.ProduceAsync("orders", "order-123", event3);
```

--------------------------------

### Install NativeAOT toolchain on Linux

Source: https://thomhurst.github.io/Dekaf/docs/testing

Installs the necessary clang and zlib dependencies required for NativeAOT compilation.

```bash
sudo apt-get update && sudo apt-get install -y clang zlib1g-dev
```

--------------------------------

### Configure Dekaf with Aspire settings

Source: https://thomhurst.github.io/Dekaf/docs/configuration/confluent-migration

Example of building configuration and registering producers and consumers in Dekaf using Aspire sections.

```csharp
using Dekaf.Extensions.DependencyInjection;



var configuration = builder.Configuration;

var ordersProducerConfig = new ConfigurationBuilder()

    .AddConfiguration(configuration.GetSection("Aspire:Confluent:Kafka:Producer:Config"))

    .AddConfiguration(configuration.GetSection("Aspire:Confluent:Kafka:Producer:orders:Config"))

    .Build();

var ordersConsumerConfig = new ConfigurationBuilder()

    .AddConfiguration(configuration.GetSection("Aspire:Confluent:Kafka:Consumer:Config"))

    .AddConfiguration(configuration.GetSection("Aspire:Confluent:Kafka:Consumer:orders:Config"))

    .Build();



builder.Services.AddDekaf(dekaf =>

{

    dekaf.AddProducerFromConfluentConfig<string, string>(

        configuration.GetSection("Aspire:Confluent:Kafka:Producer:Config"),

        producer => producer.WithBootstrapServers(

            configuration.GetConnectionString("kafka")!));



    dekaf.AddConsumerFromConfluentConfig<string, string>(

        configuration.GetSection("Aspire:Confluent:Kafka:Consumer:Config"),

        consumer => consumer

            .WithBootstrapServers(configuration.GetConnectionString("kafka")!)

            .SubscribeTo("events"));



    dekaf.AddProducerFromConfluentConfig<string, string>(

        "orders",

        ordersProducerConfig,

        producer => producer.WithBootstrapServers(

            configuration.GetConnectionString("orders")!));



    dekaf.AddConsumerFromConfluentConfig<string, string>(

        "orders",

        ordersConsumerConfig,

        consumer => consumer

            .WithBootstrapServers(configuration.GetConnectionString("orders")!)

            .SubscribeTo("orders"));

});
```

--------------------------------

### Configure Kafka Client via appsettings.json

Source: https://thomhurst.github.io/Dekaf/docs/configuration/client-dns-lookup

Example configuration for a Kafka producer using JSON settings.

```json
{

  "Kafka": {

    "Producers": {

      "Orders": {

        "BootstrapServers": "kafka.example.com:9092",

        "ClientDnsLookup": "UseAllDnsIps",

        "BootstrapResolveTimeoutMs": 120000

      }

    }

  }

}
```

--------------------------------

### Partition Reader Implementation

Source: https://thomhurst.github.io/Dekaf/docs/consumer/manual-assignment

A complete example showing how to read a specific range of messages from a Kafka partition.

```csharp
using Dekaf;



public class PartitionReader

{

    public async Task ReadPartitionAsync(

        string bootstrapServers,

        string topic,

        int partition,

        long startOffset,

        long? endOffset,

        Func<ConsumeResult<string, string>, Task> processor,

        CancellationToken ct)

    {

        await using var consumer = await Kafka.CreateConsumer<string, string>()

            .WithBootstrapServers(bootstrapServers)

            .WithAutoOffsetReset(AutoOffsetReset.None)

            .BuildAsync();



        var topicPartition = new TopicPartition(topic, partition);

        consumer.Partitions.Assign(topicPartition);

        consumer.Positions.Seek(new TopicPartitionOffset(topic, partition, startOffset));



        await foreach (var msg in consumer.ConsumeAsync(ct))

        {

            await processor(msg);



            // Stop if we've reached the end offset

            if (endOffset.HasValue && msg.Offset >= endOffset.Value)

            {

                break;

            }

        }

    }

}



// Usage: Read messages 100-200 from partition 0

var reader = new PartitionReader();

await reader.ReadPartitionAsync(

    "localhost:9092",

    "my-topic",

    partition: 0,

    startOffset: 100,

    endOffset: 200,

    async msg => Console.WriteLine(msg.Value),

    cancellationToken

);
```

--------------------------------

### Complete Secure Kafka Producer Implementation

Source: https://thomhurst.github.io/Dekaf/docs/security/sasl

A full class example showing how to inject configuration and build a secure producer.

```csharp
using Dekaf;



public class SecureKafkaClient

{

    private readonly IConfiguration _config;



    public async Task<IKafkaProducer<string, string>> CreateProducer()

    {

        return await Kafka.CreateProducer<string, string>()

            .WithBootstrapServers(_config["Kafka:BootstrapServers"])

            .UseTls()

            .WithSaslScramSha512(

                _config["Kafka:Username"],

                _config["Kafka:Password"]

            )

            .ForReliability()

            .BuildAsync();

    }

}
```

--------------------------------

### Configure Dekaf and Implement Consumer Services

Source: https://thomhurst.github.io/Dekaf/docs/dependency-injection

A comprehensive example showing service registration in Program.cs, producer injection in a controller, and a custom consumer service implementation.

```csharp
// Program.cs

var builder = WebApplication.CreateBuilder(args);



var kafkaConfig = builder.Configuration.GetSection("Kafka");



builder.Services.AddDekaf(dekaf =>

{

    // Register producer

    dekaf.AddProducer<string, Order>(producer => producer

        .WithBootstrapServers(kafkaConfig["BootstrapServers"]!)

        .WithValueSerializer(new JsonSerializer<Order>())

        .ForReliability());



    // Register consumer + background service

    dekaf.AddConsumerService<OrderProcessorService, string, Order>(consumer => consumer

        .WithBootstrapServers(kafkaConfig["BootstrapServers"]!)

        .WithGroupId(kafkaConfig["GroupId"]!)

        .WithValueDeserializer(new JsonDeserializer<Order>()));

});



var app = builder.Build();

app.MapControllers();

app.Run();



// OrderController.cs

[ApiController]

[Route("api/[controller]")]

public class OrdersController : ControllerBase

{

    private readonly IKafkaProducer<string, Order> _producer;



    public OrdersController(IKafkaProducer<string, Order> producer)

    {

        _producer = producer;

    }



    [HttpPost]

    public async Task<IActionResult> CreateOrder(Order order)

    {

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

        return Accepted();

    }

}



// OrderProcessorService.cs

public sealed class OrderProcessorService : KafkaConsumerService<string, Order>

{

    private readonly IOrderRepository _repository;



    public OrderProcessorService(

        IKafkaConsumer<string, Order> consumer,

        ILogger<OrderProcessorService> logger,

        IOrderRepository repository)

        : base(consumer, logger)

    {

        _repository = repository;

    }



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



    protected override async ValueTask ProcessAsync(

        ConsumeResult<string, Order> result, CancellationToken cancellationToken)

    {

        await _repository.SaveAsync(result.Value);

    }

}
```

--------------------------------

### Install Dekaf Compression Packages

Source: https://thomhurst.github.io/Dekaf/docs/getting-started

Add optional compression support for Kafka messages.

```bash
dotnet add package Dekaf.Compression.Lz4    # Recommended for most use cases

dotnet add package Dekaf.Compression.Zstd   # Best compression ratio

dotnet add package Dekaf.Compression.Snappy # Alternative fast codec
```

--------------------------------

### Install Dekaf via NuGet

Source: https://thomhurst.github.io/Dekaf/docs/getting-started

Use the .NET CLI to add the core Dekaf package to your project.

```bash
dotnet add package Dekaf
```

--------------------------------

### Install Vault Provider

Source: https://thomhurst.github.io/Dekaf/docs/serialization/schema-registry

Add the required NuGet package for Vault support.

```bash
dotnet add package Dekaf.SchemaRegistry.Kms.Vault
```

--------------------------------

### Assigning Partitions with Starting Offsets

Source: https://thomhurst.github.io/Dekaf/docs/consumer/manual-assignment

Use the Seek method to define specific starting offsets for manually assigned partitions.

```csharp
var first = new TopicPartition("my-topic", 0);

var second = new TopicPartition("my-topic", 1);



consumer.Partitions.Assign(first, second);

consumer.Positions.Seek(new TopicPartitionOffset("my-topic", 0, 100));  // Start at offset 100

consumer.Positions.Seek(new TopicPartitionOffset("my-topic", 1, 200));  // Start at offset 200
```

--------------------------------

### Install Dekaf JSON package

Source: https://thomhurst.github.io/Dekaf/docs/serialization/json

Use the .NET CLI to add the required serialization package to your project.

```bash
dotnet add package Dekaf.Serialization.Json
```

--------------------------------

### Run Kafka Locally with Docker

Source: https://thomhurst.github.io/Dekaf/docs/getting-started

Start a local Kafka broker instance using the Bitnami Docker image.

```bash
docker run -d --name kafka \

  -p 9092:9092 \

  -e KAFKA_CFG_NODE_ID=0 \

  -e KAFKA_CFG_PROCESS_ROLES=controller,broker \

  -e KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093 \

  -e KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT \

  -e KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=0@localhost:9093 \

  -e KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER \

  -e KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092 \

  bitnami/kafka:latest
```

--------------------------------

### Install Schema Registry Packages

Source: https://thomhurst.github.io/Dekaf/docs/serialization/schema-registry

Use these commands to add the necessary NuGet packages for Schema Registry support, including specific serializers for Avro, Protobuf, and JSON.

```bash
# Core Schema Registry support

dotnet add package Dekaf.SchemaRegistry



# For Avro serialization

dotnet add package Dekaf.SchemaRegistry.Avro



# For Protobuf serialization

dotnet add package Dekaf.SchemaRegistry.Protobuf



# Optional JSON Schema payload validation

dotnet add package Dekaf.SchemaRegistry.Json
```

--------------------------------

### Configure Batch Sizes

Source: https://thomhurst.github.io/Dekaf/docs/consumer/linq-extensions

Examples of setting batch sizes based on latency requirements and message size.

```csharp
// Small batches for low-latency processing

.Batch(10)



// Larger batches for bulk operations

.Batch(1000)



// Consider memory when batching large messages

.Batch(100)  // If messages are large
```

--------------------------------

### CoreWCF Kafka Configuration Structure

Source: https://thomhurst.github.io/Dekaf/docs/configuration/confluent-migration

Example of the existing CoreWCF Kafka configuration format.

```json
{

  "CoreWCF": {

    "Kafka": {

      "Endpoint": "net.kafka://kafka:9092/orders",

      "ConsumerConfig": {

        "GroupId": "orders-service",

        "AutoOffsetReset": "Earliest",

        "IsolationLevel": "ReadCommitted"

      },

      "ProducerConfig": {

        "Acks": "All",

        "CompressionType": "Lz4"

      }

    }

  }

}
```

--------------------------------

### Install Google Cloud KMS Provider

Source: https://thomhurst.github.io/Dekaf/docs/serialization/schema-registry

Add the required NuGet package to your project to enable Google Cloud KMS support.

```bash
dotnet add package Dekaf.SchemaRegistry.Kms.Gcp
```

--------------------------------

### Example Aspire JSON configuration

Source: https://thomhurst.github.io/Dekaf/docs/configuration/confluent-migration

A sample JSON configuration structure for Aspire Confluent Kafka producers and consumers.

```json
{

  "ConnectionStrings": {

    "kafka": "kafka:9092",

    "orders": "orders-kafka:9092"

  },

  "Aspire": {

    "Confluent": {

      "Kafka": {

        "Producer": {

          "Config": {

            "BootstrapServers": "json-default:9092",

            "ClientId": "default-producer",

            "Acks": "All",

            "MessageSendMaxRetries": 7

          },

          "orders": {

            "Config": {

              "BootstrapServers": "json-orders:9092",

              "ClientId": "orders-producer",

              "CompressionType": "Lz4"

            }

          }

        },

        "Consumer": {

          "Config": {

            "BootstrapServers": "json-default:9092",

            "ClientId": "default-consumer",

            "GroupId": "default-group",

            "AutoOffsetReset": "Latest",

            "FetchMinBytes": 4096

          },

          "orders": {

            "Config": {

              "BootstrapServers": "json-orders:9092",

              "ClientId": "orders-consumer",

              "GroupId": "orders-group",

              "AutoOffsetReset": "Earliest"

            }

          }

        }

      }

    }

  }

}
```

--------------------------------

### Install Dekaf Dependency Injection package

Source: https://thomhurst.github.io/Dekaf/docs/configuration/confluent-migration

Add the required NuGet package to your project to enable configuration migration.

```bash
dotnet add package Dekaf.Extensions.DependencyInjection
```

--------------------------------

### Implement Exactly-Once Processing Loop

Source: https://thomhurst.github.io/Dekaf/docs/producer/transactions

A complete example demonstrating how to consume, transform, and produce messages within a transaction, including offset commitment.

```csharp
public class ExactlyOnceProcessor

{

    private readonly IKafkaConsumer<string, string> _consumer;

    private readonly IKafkaProducer<string, string> _producer;

    private readonly ILogger _logger;



    public async Task RunAsync(CancellationToken ct)

    {

        await _producer.InitTransactionsAsync();



        await foreach (var batch in _consumer.ConsumeAsync(ct).Batch(100))

        {

            try

            {

                await _producer.BeginTransactionAsync();



                var offsets = new List<TopicPartitionOffset>();



                foreach (var msg in batch)

                {

                    var result = Transform(msg.Value);

                    await _producer.ProduceAsync("output", msg.Key, result);

                    offsets.Add(new TopicPartitionOffset(msg.Topic, msg.Partition, msg.Offset + 1));

                }



                await _producer.SendOffsetsToTransactionAsync(

                    _consumer.ConsumerGroupMetadata!,

                    offsets

                );



                await _producer.CommitTransactionAsync();

                _logger.LogInformation("Committed batch of {Count} messages", batch.Count);

            }

            catch (Exception ex)

            {

                _logger.LogError(ex, "Transaction failed, aborting");

                await _producer.AbortTransactionAsync();

            }

        }

    }



    private string Transform(string input) => input.ToUpperInvariant();

}
```

--------------------------------

### Install Azure Key Vault Provider

Source: https://thomhurst.github.io/Dekaf/docs/serialization/schema-registry

Command to add the required NuGet package for Azure Key Vault support.

```bash
dotnet add package Dekaf.SchemaRegistry.Kms.Azure
```

--------------------------------

### Configure Azure Key Vault KMS Provider

Source: https://thomhurst.github.io/Dekaf/docs/serialization/schema-registry

Example showing how to initialize the AzureKeyVaultKmsProvider with default credentials and register it with the Schema Registry rule handler.

```csharp
using Azure.Identity;

using Dekaf.SchemaRegistry;

using Dekaf.SchemaRegistry.Kms.Azure;



var credential = new DefaultAzureCredential();

var azureKms = new AzureKeyVaultKmsProvider(credential);

var confluentAzureKms = new AzureKeyVaultKmsProvider(

    credential,

    type: AzureKeyVaultKmsProvider.ConfluentType);

var csfle = new SchemaRegistryCsfleRuleHandler(

    schemaRegistry,

    [azureKms, confluentAzureKms]);
```

--------------------------------

### Demonstrate Message Ordering with Keys

Source: https://thomhurst.github.io/Dekaf/docs/producer/partitioning

Illustrates how using the same key ensures sequential processing within a partition, while different keys may result in out-of-order consumption.

```csharp
// ✅ Guaranteed order - same key = same partition

await producer.ProduceAsync("orders", "order-123", "created");

await producer.ProduceAsync("orders", "order-123", "paid");

await producer.ProduceAsync("orders", "order-123", "shipped");

// Consumer will see: created -> paid -> shipped



// ⚠️ No ordering guarantee - different keys may go to different partitions

await producer.ProduceAsync("orders", "order-1", "created");

await producer.ProduceAsync("orders", "order-2", "created");

await producer.ProduceAsync("orders", "order-1", "paid");

// order-2's "created" might be consumed before order-1's "paid"
```

--------------------------------

### Guid Serializer Usage

Source: https://thomhurst.github.io/Dekaf/docs/serialization/built-in

Serializes Guids into a 16-byte binary representation.

```csharp
using Dekaf;



var producer = await Kafka.CreateProducer<Guid, string>()

    .WithBootstrapServers("localhost:9092")

    .BuildAsync();



await producer.ProduceAsync("topic", Guid.NewGuid(), "value");
```

--------------------------------

### Direct use of in-memory Kafka clients

Source: https://thomhurst.github.io/Dekaf/docs/testing

Demonstrates setting up an in-memory cluster, producer, and consumer to perform basic produce and consume operations.

```csharp
using Dekaf.Consumer;

using Dekaf.Producer;

using Dekaf.Testing;



var cluster = new InMemoryKafkaCluster();

cluster.CreateTopic("orders", partitionCount: 3);



var producer = new InMemoryProducer<string, string>(cluster);

var consumer = new InMemoryConsumer<string, string>(

    cluster,

    new InMemoryConsumerOptions

    {

        GroupId = "orders-service",

        AutoOffsetReset = AutoOffsetReset.Earliest

    });



await producer.ProduceAsync(new ProducerMessage<string, string>

{

    Topic = "orders",

    Key = "order-1",

    Value = "created"

});



consumer.Subscribe("orders");

var record = await consumer.ConsumeOneAsync(TimeSpan.FromSeconds(1));
```

--------------------------------

### Enable Compression via Convenience Methods

Source: https://thomhurst.github.io/Dekaf/docs/compression

Configure compression directly on the producer builder using fluent extension methods.

```csharp
using Dekaf;



var producer = await Kafka.CreateProducer<string, string>()

    .WithBootstrapServers("localhost:9092")

    .UseLz4Compression()    // or UseZstdCompression(), UseSnappyCompression(), UseGzipCompression()

    .BuildAsync();
```

--------------------------------

### Configure Confluent Cloud

Source: https://thomhurst.github.io/Dekaf/docs/security/tls

Standard TLS setup combined with SASL Plain authentication.

```csharp
using Dekaf;



var producer = await Kafka.CreateProducer<string, string>()

    .WithBootstrapServers("pkc-xxxxx.us-east-1.aws.confluent.cloud:9092")

    .UseTls()

    .WithSaslPlain(apiKey, apiSecret)

    .BuildAsync();
```

--------------------------------

### Implement Zero-Allocation Serializer

Source: https://thomhurst.github.io/Dekaf/docs/serialization/custom

Example of using IBufferWriter for efficient, zero-allocation serialization of integers.

```csharp
public class Int32Serializer : ISerializer<int>, IDeserializer<int>

{

    public void Serialize(int value, IBufferWriter<byte> output, SerializationContext context)

    {

        var span = output.GetSpan(4);

        BinaryPrimitives.WriteInt32BigEndian(span, value);

        output.Advance(4);

    }



    public int Deserialize(ReadOnlySequence<byte> data, SerializationContext context)

    {

        Span<byte> buffer = stackalloc byte[4];

        data.Slice(0, 4).CopyTo(buffer);

        return BinaryPrimitives.ReadInt32BigEndian(buffer);

    }

}
```

--------------------------------

### Configure Producer via Dependency Injection

Source: https://thomhurst.github.io/Dekaf/docs/configuration/producer-options

Demonstrates basic and advanced producer configuration using the fluent builder within DI registration.

```csharp
// Before: DI examples only showed the small common subset.

builder.Services.AddDekaf(dekaf =>

{

    dekaf.AddProducer<string, string>(producer => producer

        .WithBootstrapServers("localhost:9092"));

});



// After: DI uses the full producer builder.

builder.Services.AddDekaf(dekaf =>

{

    dekaf.AddProducer<string, string>(producer => producer

        .WithBootstrapServers("localhost:9092")

        .WithLinger(TimeSpan.FromMilliseconds(5))

        .WithBatchSize(64 * 1024)

        .WithBufferMemory(256 * 1024 * 1024)

        .WithSaslScramSha512("user", "password")

        .ForHighThroughput());

});
```

--------------------------------

### Define and Use Custom Presets

Source: https://thomhurst.github.io/Dekaf/docs/configuration/presets

Shows how to create extension methods for ProducerBuilder and ConsumerBuilder to enforce consistent settings across an application.

```csharp
using Dekaf;



public static class DekafPresets

{

    public static ProducerBuilder<TKey, TValue> ForOrderProcessing<TKey, TValue>(

        this ProducerBuilder<TKey, TValue> builder)

    {

        return builder

            .ForReliability()

            .UseLz4Compression()

            .WithLingerMs(1);  // Slight batching for efficiency

    }



    public static ConsumerBuilder<TKey, TValue> ForOrderProcessing<TKey, TValue>(

        this ConsumerBuilder<TKey, TValue> builder)

    {

        return builder

            .WithOffsetCommitMode(OffsetCommitMode.Manual)

            .WithAutoOffsetReset(AutoOffsetReset.Earliest)

            .ForLowLatency();

    }

}



// Usage

var producer = await Kafka.CreateProducer<string, Order>()

    .WithBootstrapServers("localhost:9092")

    .ForOrderProcessing()

    .BuildAsync();
```

--------------------------------

### Configure Consumer via Dependency Injection

Source: https://thomhurst.github.io/Dekaf/docs/configuration/consumer-options

Demonstrates basic and advanced consumer configuration using the fluent builder pattern within the AddDekaf registration.

```csharp
// Before: DI examples only showed connection and group settings.

builder.Services.AddDekaf(dekaf =>

{

    dekaf.AddConsumer<string, string>(consumer => consumer

        .WithBootstrapServers("localhost:9092")

        .WithGroupId("orders"));

});



// After: DI uses the full consumer builder.

builder.Services.AddDekaf(dekaf =>

{

    dekaf.AddConsumer<string, string>(consumer => consumer

        .WithBootstrapServers("localhost:9092")

        .WithGroupId("orders")

        .WithFetchMinBytes(1024)

        .WithFetchMaxBytes(50 * 1024 * 1024)

        .WithPrefetchPipelineDepth(4)

        .WithSaslScramSha512("user", "password")

        .SubscribeTo("orders"));

});
```

--------------------------------

### Troubleshooting Certificate Validation

Source: https://thomhurst.github.io/Dekaf/docs/security/tls

Examples for handling specific validation errors or debugging certificate issues.

```text
AuthenticationException: The remote certificate was rejected
```

```csharp
var tlsConfig = new TlsConfig

{

    CaCertificatePath = "/path/to/ca.crt",

    ValidateServerCertificate = true,

    ValidateServerCertificateHostName = false

};
```

```csharp
// ⚠️ NEVER use in production

var tlsConfig = new TlsConfig

{

    ValidateServerCertificate = false

};
```

```csharp
var tlsConfig = new TlsConfig

{

    CaCertificatePath = "/path/to/ca.crt",

    ServerCertificateValidationCallback = (sender, cert, chain, errors) =>

    {

        Console.WriteLine($"Subject: {cert.Subject}");

        Console.WriteLine($"Issuer: {cert.Issuer}");

        Console.WriteLine($"Errors: {errors}");

        return errors == SslPolicyErrors.None;

    }

};
```

--------------------------------

### Bind Consumer Settings from Configuration

Source: https://thomhurst.github.io/Dekaf/docs/configuration/consumer-options

Shows the JSON structure for consumer settings and how to apply them using GetSection during service registration.

```json
{

  "Kafka": {

    "Consumers": {

      "Orders": {

        "BootstrapServers": [

          "broker1:9092",

          "broker2:9092"

        ],

        "ClientId": "orders-consumer",

        "GroupId": "orders",

        "AutoOffsetReset": "Earliest",

        "OffsetCommitMode": "Manual",

        "FetchMinBytes": 1024,

        "FetchMaxBytes": 52428800,

        "FetchBufferMemoryBytes": 104857600,

        "FetchMaxWaitMs": 200,

        "MaxPollRecords": 500,

        "UseTls": true

      }

    }

  }

}
```

```csharp
builder.Services.AddDekaf(dekaf =>

{

    dekaf.AddConsumer<string, Order>(

        builder.Configuration.GetSection("Kafka:Consumers:Orders"),

        consumer => consumer

            .WithValueDeserializer(new JsonDeserializer<Order>())

            .SubscribeTo("orders"));

});
```

--------------------------------

### Configure Connection Timeout

Source: https://thomhurst.github.io/Dekaf/docs/configuration/consumer-options

Sets the maximum time allowed for socket connection setup, including handshakes.

```csharp
.WithConnectionTimeout(TimeSpan.FromSeconds(10))
```

```csharp
.WithConnectionTimeout(TimeSpan.FromSeconds(10))

.WithConnectionTimeoutMax(TimeSpan.FromSeconds(127))
```

--------------------------------

### Configure AWS MSK with IAM

Source: https://thomhurst.github.io/Dekaf/docs/security/tls

Standard TLS setup combined with AWS MSK IAM authentication.

```csharp
using Dekaf;



var producer = await Kafka.CreateProducer<string, string>()

    .WithBootstrapServers("broker1.msk.us-east-1.amazonaws.com:9098")

    .UseTls()

    .WithAwsMskIam()

    .BuildAsync();
```

--------------------------------

### Configure Producer with Security and Presets

Source: https://thomhurst.github.io/Dekaf/docs/configuration/presets

Demonstrates chaining security settings like TLS and SASL with a reliability preset.

```csharp
using Dekaf;



var producer = await Kafka.CreateProducer<string, string>()

    .WithBootstrapServers("kafka.example.com:9093")

    .UseTls()

    .WithSaslScramSha512("username", "password")

    .ForReliability()  // Performance preset

    .BuildAsync();
```

--------------------------------

### Configure Dekaf DI Builder

Source: https://thomhurst.github.io/Dekaf/docs/dependency-injection

Demonstrates the transition from basic registration to a full builder configuration including retry policies and advanced settings.

```csharp
builder.Services.AddDekaf(dekaf =>

{

    dekaf.AddProducer<string, string>(producer => producer

        .WithBootstrapServers("localhost:9092")

        .WithAcks(Acks.All));

});
```

```csharp
var retryPolicy = new FixedDelayRetryPolicy

{

    Delay = TimeSpan.FromMilliseconds(100),

    MaxAttempts = 3

};



builder.Services.AddDekaf(dekaf =>

{

    dekaf.AddProducer<string, string>(producer => producer

        .WithBootstrapServers("localhost:9092")

        .WithLinger(TimeSpan.FromMilliseconds(5))

        .WithBatchSize(64 * 1024)

        .WithBufferMemory(256 * 1024 * 1024)

        .WithSaslScramSha512("user", "password")

        .WithRetryPolicy(retryPolicy)

        .ForHighThroughput());



    dekaf.AddConsumer<string, string>(consumer => consumer

        .WithBootstrapServers("localhost:9092")

        .WithGroupId("orders")

        .WithFetchMinBytes(1024)

        .WithFetchMaxBytes(50 * 1024 * 1024)

        .WithPrefetchPipelineDepth(4)

        .WithRetryPolicy(retryPolicy)

        .SubscribeTo("orders"));

});
```

--------------------------------

### Compare Batch and Sequential Production

Source: https://thomhurst.github.io/Dekaf/docs/producer/batch-production

Demonstrates the performance difference between sequential production and batch production.

```csharp
// ❌ Inefficient - waits for each message

foreach (var msg in messages)

{

    await producer.ProduceAsync("topic", msg.Key, msg.Value);

}



// ✅ Efficient - sends all at once

await producer.ProduceAllAsync(messages);
```

--------------------------------

### Configuring Keyed Clients with Sections

Source: https://thomhurst.github.io/Dekaf/docs/dependency-injection

Register keyed producers and consumers using configuration sections.

```csharp
builder.Services.AddDekaf(dekaf =>
{
    dekaf.AddProducer<string, Order>(
        "orders",
        builder.Configuration.GetSection("Kafka:Producers:Orders"),
        producer => producer.WithValueSerializer(new JsonSerializer<Order>()));

    dekaf.AddConsumer<string, Order>(
        "orders",
        builder.Configuration.GetSection("Kafka:Consumers:Orders"),
        consumer => consumer
            .WithValueDeserializer(new JsonDeserializer<Order>())
            .SubscribeTo("orders"));
});
```

--------------------------------

### Configure Bootstrap Servers

Source: https://thomhurst.github.io/Dekaf/docs/configuration/producer-options

Sets the initial Kafka broker addresses. Use the params overload for multiple servers or the single string for comma-separated configuration values.

```csharp
// Single server

.WithBootstrapServers("localhost:9092")



// Multiple servers (typed params)

.WithBootstrapServers("broker1:9092", "broker2:9092", "broker3:9092")



// Convenience for comma-separated configuration values

.WithBootstrapServers("broker1:9092,broker2:9092,broker3:9092")
```

--------------------------------

### Configure Auto Offset Reset via JSON

Source: https://thomhurst.github.io/Dekaf/docs/configuration/consumer-options

Configuration examples for setting AutoOffsetReset using JSON binding.

```json
{

  "AutoOffsetReset": "ByDuration",

  "AutoOffsetResetDuration": "24:00:00"

}
```

```json
{

  "AutoOffsetReset": "by_duration:PT24H"

}
```

--------------------------------

### Create Headers Quickly

Source: https://thomhurst.github.io/Dekaf/docs/producer/headers

Initialize a header collection with a single key-value pair using the shorthand creation method.

```csharp
var headers = Headers.Create("trace-id", traceId);
```

--------------------------------

### Configure producer with sticky partitioning and linger

Source: https://thomhurst.github.io/Dekaf/docs/producer/partitioning

Initializes a producer with sticky partitioning and a linger duration to encourage batching.

```csharp
using Dekaf;



var producer = await Kafka.CreateProducer<string, string>()

    .WithBootstrapServers("localhost:9092")

    .WithPartitioner(PartitionerType.Sticky)

    .WithLingerMs(5)

    .BuildAsync();



// These will likely batch together in one partition

await producer.FireAsync("events", null, "event1");

await producer.FireAsync("events", null, "event2");

await producer.FireAsync("events", null, "event3");
```

--------------------------------

### Compare Regular and Topic Producer Usage

Source: https://thomhurst.github.io/Dekaf/docs/producer/topic-producer

Demonstrates the difference between specifying a topic on every call versus using an implicit topic producer.

```csharp
// Regular producer - topic on every call

await producer.ProduceAsync("orders", "key", "value");

await producer.ProduceAsync("orders", "key2", "value2");
```

```csharp
// Topic producer - no topic parameter

await producer.ProduceAsync("key", "value");

await producer.ProduceAsync("key2", "value2");
```

--------------------------------

### Configure OrderProcessorService with Retry Policy

Source: https://thomhurst.github.io/Dekaf/docs/hosted-services

Example of implementing a KafkaConsumerService with a fixed delay retry policy passed to the base constructor.

```csharp
public sealed class OrderProcessorService : KafkaConsumerService<string, Order>

{

    public OrderProcessorService(

        IKafkaConsumer<string, Order> consumer,

        ILogger<OrderProcessorService> logger,

        DeadLetterOptions deadLetterOptions)   // resolved from DI when configured

        : base(

            consumer,

            logger,

            deadLetterOptions,

            retryPolicy: new FixedDelayRetryPolicy

            {

                Delay = TimeSpan.FromMilliseconds(200),

                MaxAttempts = 3

            })

    {

    }

    // ...

}
```

--------------------------------

### Incorrect Usage of Filtering Operators

Source: https://thomhurst.github.io/Dekaf/docs/consumer/linq-extensions

Example demonstrating data loss when using filtering operators to defer processing of messages.

```csharp
// ❌ DATA LOSS - messages skipped while unhealthy are dropped forever

await foreach (var msg in consumer.ConsumeAsync(ct)

    .Where(_ => downstream.IsHealthy))

{

    await ProcessAsync(msg);

    consumer.StoreOffset(msg);

    await consumer.CommitAsync(); // commits past every skipped message

}
```

--------------------------------

### Bind Producer Configuration from JSON

Source: https://thomhurst.github.io/Dekaf/docs/configuration/producer-options

Shows the structure of a JSON configuration section for producers and how to bind it using GetSection.

```json
{

  "Kafka": {

    "Producers": {

      "Orders": {

        "BootstrapServers": "broker1:9092,broker2:9092",

        "ClientId": "orders-producer",

        "Acks": "All",

        "LingerMs": 5,

        "BatchSize": 65536,

        "CompressionType": "Lz4",

        "EnableIdempotence": true,

        "UseTls": true,

        "SaslMechanism": "ScramSha512",

        "SaslUsername": "user",

        "SaslPassword": "password"

      }

    }

  }

}
```

```csharp
builder.Services.AddDekaf(dekaf =>

{

    dekaf.AddProducer<string, Order>(

        builder.Configuration.GetSection("Kafka:Producers:Orders"),

        producer => producer.WithValueSerializer(new JsonSerializer<Order>()));

});
```

--------------------------------

### Run Capability Benchmarks

Source: https://thomhurst.github.io/Dekaf/docs/testing

Executes benchmarks for capability creation and lookup in Release configuration.

```bash
dotnet run --configuration Release \

  --project tools/Dekaf.Benchmarks/Dekaf.Benchmarks.csproj -- \

  --filter "*ApiVersionNegotiationBenchmarks*"
```

--------------------------------

### Configure Confluent Cloud Producer with SASL PLAIN

Source: https://thomhurst.github.io/Dekaf/docs/security/sasl

Sets up a producer for Confluent Cloud using environment variables for API credentials.

```csharp
using Dekaf;



var apiKey = Environment.GetEnvironmentVariable("CONFLUENT_API_KEY");

var apiSecret = Environment.GetEnvironmentVariable("CONFLUENT_API_SECRET");



var producer = await Kafka.CreateProducer<string, string>()

    .WithBootstrapServers("pkc-xxxxx.us-east-1.aws.confluent.cloud:9092")

    .UseTls()

    .WithSaslPlain(apiKey, apiSecret)

    .BuildAsync();
```

--------------------------------

### Explicit Partition Assignment Example

Source: https://thomhurst.github.io/Dekaf/docs/producer/partitioning

Directly specify the target partition for a message. Note that this creates a dependency on the topic's partition count.

```csharp
var message = new ProducerMessage<string, string>

{

    Topic = "events",

    Partition = 0,  // Always send to partition 0

    Key = "key",

    Value = "value"

};



await producer.ProduceAsync(message);
```

--------------------------------

### Compare FireAsync and ProduceAsync

Source: https://thomhurst.github.io/Dekaf/docs/producer/fire-and-forget

Demonstrates the difference between enqueuing locally with FireAsync and waiting for broker acknowledgment with ProduceAsync.

```csharp
// Enqueues locally; does not wait for broker acknowledgment

await producer.FireAsync("my-topic", "key", "value");
```

```csharp
// Waits for broker acknowledgment and returns metadata

await producer.ProduceAsync("my-topic", "key", "value");
```

--------------------------------

### Chain LINQ Extensions for Pipelines

Source: https://thomhurst.github.io/Dekaf/docs/consumer/linq-extensions

Examples of chaining LINQ operators to filter, transform, and batch messages, or to sample specific message ranges.

```csharp
// Filter, transform, batch

await foreach (var batch in consumer.ConsumeAsync(ct)

    .Where(m => m.Key != null)

    .Where(m => m.Value.Length > 0)

    .Select(m => new ProcessedMessage(m.Key!, m.Value, m.Timestamp))

    .Batch(50))

{

    await ProcessBatchAsync(batch);

    await consumer.CommitAsync();
}
```

```csharp
// Skip old messages, take a sample

var recentMessages = new List<ConsumeResult<string, string>>();

var cutoff = DateTimeOffset.UtcNow.AddHours(-1);



await foreach (var msg in consumer.ConsumeAsync(ct)

    .SkipWhile(m => m.Timestamp < cutoff)

    .Take(1000))

{

    recentMessages.Add(msg);

}
```

--------------------------------

### Configure Auto Offset Reset

Source: https://thomhurst.github.io/Dekaf/docs/consumer/basics

Sets the starting position for consumers without committed offsets using either earliest, latest, or duration-based strategies.

```csharp
using Dekaf;



var consumer = await Kafka.CreateConsumer<string, string>()

    .WithBootstrapServers("localhost:9092")

    .WithGroupId("my-group")

    .WithAutoOffsetReset(AutoOffsetReset.Earliest)  // Start from beginning

    .BuildAsync();



var recentConsumer = await Kafka.CreateConsumer<string, string>()

    .WithBootstrapServers("localhost:9092")

    .WithGroupId("recent-only")

    .WithAutoOffsetResetByDuration(TimeSpan.FromHours(24))

    .BuildAsync();
```

--------------------------------

### Configure Two-Phase Commit Participation

Source: https://thomhurst.github.io/Dekaf/docs/producer/transactions

Initializes a producer with two-phase commit enabled and demonstrates the preparation and completion workflow.

```csharp
await using var producer = await Kafka.CreateProducer<string, string>()

    .WithBootstrapServers("localhost:9092")

    .WithTransactionalId("orders-2pc")

    .WithTwoPhaseCommit()

    .BuildAsync();



await producer.InitTransactionsAsync();



await using var transaction = producer.BeginTransaction();

await transaction.ProduceAsync(new ProducerMessage<string, string>

{

    Topic = "orders",

    Key = orderId,

    Value = orderJson

});



var prepared = await transaction.PrepareAsync();



// Store prepared.ToString() with the external transaction decision.

await producer.CompletePreparedTransactionAsync(prepared, committed: true);
```

--------------------------------

### Consume decompressed messages with Dekaf

Source: https://thomhurst.github.io/Dekaf/docs/compression

Consumers automatically handle decompression. Ensure the required compression codec package is installed in the consumer application to avoid failures.

```csharp
using Dekaf;



// Consumer handles decompression automatically

var consumer = await Kafka.CreateConsumer<string, string>()

    .WithBootstrapServers("localhost:9092")

    .WithGroupId("my-group")

    .BuildAsync();



await foreach (var msg in consumer.ConsumeAsync(ct))

{

    // msg.Value is already decompressed

    Console.WriteLine(msg.Value);

}
```

--------------------------------

### Dependency Injection with Topic Producers

Source: https://thomhurst.github.io/Dekaf/docs/producer/topic-producer

Shows how to register an ITopicProducer in a service collection and inject it into a service for producing messages to a specific topic.

```csharp
using Dekaf;



// Registration

services.AddSingleton<ITopicProducer<string, OrderEvent>>(sp =>

{

    return Kafka.CreateProducer<string, OrderEvent>()

        .WithBootstrapServers(config["Kafka:BootstrapServers"])

        .WithValueSerializer(new JsonSerializer<OrderEvent>())

        .BuildForTopic("orders");

});



// Usage

public class OrderService

{

    private readonly ITopicProducer<string, OrderEvent> _producer;



    public OrderService(ITopicProducer<string, OrderEvent> producer)

    {

        _producer = producer;

    }



    public async Task PlaceOrderAsync(Order order)

    {

        var @event = new OrderEvent { OrderId = order.Id, Status = "Placed" };

        await _producer.ProduceAsync(order.Id, @event);

    }

}
```

--------------------------------

### Create a topic producer with configuration

Source: https://thomhurst.github.io/Dekaf/docs/producer/topic-producer

Configures a topic producer with additional settings like acknowledgments and idempotence.

```csharp
using Dekaf;



await using var producer = await Kafka.CreateProducer<string, string>()

    .WithBootstrapServers("localhost:9092")

    .WithAcks(Acks.All)

    .EnableIdempotence()

    .BuildForTopicAsync("orders");



await producer.ProduceAsync("order-123", orderJson);
```

--------------------------------

### Identify messages for deletion in custom stores

Source: https://thomhurst.github.io/Dekaf/docs/producer/outbox

Demonstrates three strategies for identifying messages to delete in MarkPublishedAsync, including relational IDs, unique GUIDs, or subclassing for native NoSQL IDs.

```csharp
// 1. Relational: by the Id column (what the EF Core store does)

var ids = published.Select(m => m.Id);



// 2. Any store: by MessageId, the always-present unique GUID

var ids = published.Select(m => m.MessageId);



// 3. NoSQL with native ids: subclass OutboxMessage, return your subclass

//    from GetNextBatchAsync, and downcast on the way back

private sealed class MongoOutboxMessage : OutboxMessage

{

    public required ObjectId DocumentId { get; init; }

}



public ValueTask MarkPublishedAsync(int bucket, IReadOnlyList<OutboxMessage> published, CancellationToken ct)

{

    var documentIds = published.Cast<MongoOutboxMessage>().Select(m => m.DocumentId);

    // collection.DeleteManyAsync(filter on documentIds)

}
```

--------------------------------

### Implement a Kafka Producer and Consumer

Source: https://thomhurst.github.io/Dekaf/docs/getting-started

Demonstrates a complete workflow running a consumer in a background task and producing messages to a topic.

```csharp
using Dekaf;

using Dekaf.Producer;

using Dekaf.Consumer;



const string bootstrapServers = "localhost:9092";

const string topic = "getting-started";



// Start the consumer in the background

var cts = new CancellationTokenSource();

var consumerTask = Task.Run(async ()

{

    await using var consumer = await Kafka.CreateConsumer<string, string>()

        .WithBootstrapServers(bootstrapServers)

        .WithGroupId("getting-started-group")

        .WithAutoOffsetReset(AutoOffsetReset.Earliest)

        .SubscribeTo(topic)

        .BuildAsync();



    await foreach (var msg in consumer.ConsumeAsync(cts.Token))

    {

        Console.WriteLine($"[Consumer] {msg.Key}: {msg.Value}");

    }

});



// Give the consumer time to join the group

await Task.Delay(2000);



// Produce some messages

await using var producer = await Kafka.CreateProducer<string, string>()

    .WithBootstrapServers(bootstrapServers)

    .BuildAsync();



for (int i = 1; i <= 5; i++)

{

    var key = $"message-{i}";

    var value = $"Hello from message {i}!";



    await producer.ProduceAsync(topic, key, value);

    Console.WriteLine($"[Producer] Sent: {key}");

}



// Wait for messages to be consumed

await Task.Delay(2000);

cts.Cancel();



try { await consumerTask; }

catch (OperationCanceledException) { }



Console.WriteLine("Done!");
```

--------------------------------

### Register a custom outbox store

Source: https://thomhurst.github.io/Dekaf/docs/producer/outbox

Shows how to register a custom IOutboxStore implementation and configure the Dekaf outbox relay in the dependency injection container.

```csharp
builder.Services.AddSingleton<IOutboxStore, DynamoDbOutboxStore>();

builder.AddDekafOutboxRelay(

    producer => producer.WithBootstrapServers("localhost:9092"));
```

--------------------------------

### Initialize GcpKmsProvider

Source: https://thomhurst.github.io/Dekaf/docs/serialization/schema-registry

Register the provider with the SchemaRegistryCsfleRuleHandler to enable field-level encryption.

```csharp
using Dekaf.SchemaRegistry;
using Dekaf.SchemaRegistry.Kms.Gcp;

var gcpKms = new GcpKmsProvider();
var csfle = new SchemaRegistryCsfleRuleHandler(schemaRegistry, [gcpKms]);
```

--------------------------------

### Seek to Offsets

Source: https://thomhurst.github.io/Dekaf/docs/consumer/manual-assignment

Demonstrates how to assign a partition and seek to the beginning, end, or a specific offset.

```csharp
consumer.Partitions.Assign(new TopicPartition("my-topic", 0));



// Seek to beginning

consumer.Positions.SeekToBeginning(new TopicPartition("my-topic", 0));



// Seek to end

consumer.Positions.SeekToEnd(new TopicPartition("my-topic", 0));



// Seek to specific offset

consumer.Positions.Seek(new TopicPartitionOffset("my-topic", 0, 12345));
```

--------------------------------

### Configure Bootstrap Servers

Source: https://thomhurst.github.io/Dekaf/docs/configuration/consumer-options

Sets the Kafka broker addresses. Prefer the typed string array overload for code-based configuration.

```csharp
.WithBootstrapServers("localhost:9092")

.WithBootstrapServers("broker1:9092", "broker2:9092")

.WithBootstrapServers("broker1:9092,broker2:9092")
```

--------------------------------

### Fluent Configuration for NativeAOT

Source: https://thomhurst.github.io/Dekaf/docs/dependency-injection

Use fluent configuration to register producers, consumers, and admin clients to avoid dynamic binding issues with IConfiguration.

```csharp
builder.Services.AddDekaf(dekaf =>

{

    dekaf.AddProducer<string, Order>(

        new ProducerOptions

        {

            BootstrapServers = ["broker1:9092", "broker2:9092"],

            ClientId = "orders-producer"

        },

        producer => producer.WithValueSerializer(new JsonSerializer<Order>()));



    dekaf.AddConsumer<string, Order>(

        new ConsumerOptions

        {

            BootstrapServers = ["broker1:9092", "broker2:9092"],

            ClientId = "orders-consumer",

            GroupId = "orders"

        },

        consumer => consumer

            .WithValueDeserializer(new JsonDeserializer<Order>())

            .SubscribeTo("orders"));



    dekaf.AddAdminClient(

        new AdminClientOptions

        {

            BootstrapServers = ["broker1:9092", "broker2:9092"],

            ClientId = "orders-admin"

        });

});
```

--------------------------------

### Manage Memory with ArrayPool

Source: https://thomhurst.github.io/Dekaf/docs/performance

Demonstrates renting and returning byte arrays to minimize garbage collection pressure.

```csharp
// Buffers are rented from the pool, not allocated

var buffer = ArrayPool<byte>.Shared.Rent(minSize);

try

{

    // Use buffer

}

finally

{

    ArrayPool<byte>.Shared.Return(buffer);

}
```

--------------------------------

### Configure Adaptive Connection Scaling

Source: https://thomhurst.github.io/Dekaf/docs/performance

Shows how to enable, customize, or disable adaptive connection scaling for producers.

```csharp
// Adaptive scaling is enabled by default — just build the producer

var producer = Kafka.CreateProducer<string, string>()

    .WithBootstrapServers("localhost:9092")

    .Build();



// Customize the maximum connections

var producer = Kafka.CreateProducer<string, string>()

    .WithBootstrapServers("localhost:9092")

    .WithAdaptiveConnections(maxConnections: 5)

    .Build();



// Disable if you need fixed connection topology

var producer = Kafka.CreateProducer<string, string>()

    .WithBootstrapServers("localhost:9092")

    .WithoutAdaptiveConnections()

    .WithConnectionsPerBroker(3)

    .Build();
```

--------------------------------

### Create Multiple Independent Consumer Groups

Source: https://thomhurst.github.io/Dekaf/docs/consumer/consumer-groups

Demonstrates how to instantiate multiple consumers with different group IDs to process the same topic independently.

```csharp
using Dekaf;



// Analytics group - processes all messages

var analyticsConsumer = await Kafka.CreateConsumer<string, string>()

    .WithGroupId("analytics")

    .SubscribeTo("orders")

    .BuildAsync();



// Notification group - also processes all messages

var notificationConsumer = await Kafka.CreateConsumer<string, string>()

    .WithGroupId("notifications")

    .SubscribeTo("orders")

    .BuildAsync();
```

--------------------------------

### Implement Rebalance Listener in C#

Source: https://thomhurst.github.io/Dekaf/docs/consumer/consumer-groups

Implement IRebalanceListener and IPartitionStopListener to handle partition assignment, revocation, loss, and graceful shutdown events. Register the listener using the WithRebalanceListener method when building the consumer.

```csharp
using Dekaf;

public sealed class MyRebalanceListener : IRebalanceListener, IPartitionStopListener
{
    public ValueTask OnPartitionsAssignedAsync(
        IEnumerable<TopicPartition> partitions,
        CancellationToken ct)
    {
        Console.WriteLine($"Assigned: {string.Join(", ", partitions)}");
        // Initialize resources for these partitions
        return ValueTask.CompletedTask;
    }

    public ValueTask OnPartitionsRevokedAsync(
        IEnumerable<TopicPartition> partitions,
        CancellationToken ct)
    {
        Console.WriteLine($"Revoked: {string.Join(", ", partitions)}");
        // Commit completed offsets, clean up resources
        return ValueTask.CompletedTask;
    }

    public ValueTask OnPartitionsLostAsync(
        IEnumerable<TopicPartition> partitions,
        CancellationToken ct)
    {
        Console.WriteLine($"Lost: {string.Join(", ", partitions)}");
        // Partitions were taken away. Do not commit offsets here.
        return ValueTask.CompletedTask;
    }

    public ValueTask OnPartitionsStoppedAsync(
        IEnumerable<TopicPartition> partitions,
        CancellationToken ct)
    {
        Console.WriteLine($"Stopped: {string.Join(", ", partitions)}");
        // Normal shutdown: drain and dispose partition-scoped resources
        return ValueTask.CompletedTask;
    }
}

var consumer = await Kafka.CreateConsumer<string, string>()
    .WithBootstrapServers("localhost:9092")
    .WithGroupId("my-group")
    .WithRebalanceListener(new MyRebalanceListener())
    .BuildAsync();
```

--------------------------------

### Initialize a Kafka Producer

Source: https://thomhurst.github.io/Dekaf/docs/getting-started

Create a producer instance using the Dekaf namespace and bootstrap servers.

```csharp
using Dekaf;



var producer = await Kafka.CreateProducer<string, string>()

    .WithBootstrapServers("localhost:9092")

    .BuildAsync();
```
Tokens:
Raw