Kafka in .NET with Confluent.Kafka: Producers, Consumer Groups, Offset Commits and Rebalancing

Version note: The code in this post was built and run with .NET 10, Apache Kafka 4.2.1 (KRaft) and Confluent.Kafka 2.16.0.

Guess First

A Kafka consumer runs with the default settings; the only difference is that the auto commit interval is lowered to 1 second. Processing each message takes 700 ms. The process crashes while the consumer is processing offset 3 of partition 1. When the consumer group restarts, which offset does it continue reading partition 1 from: 3 or 4? In other words, is the half-finished invoice issued again, or never issued at all?

Second question: you send 100 messages with Produce and dispose the producer at the end of a using block. How many messages reach the broker?

Kafka Has No Queue, It Has a Log

In RabbitMQ and Azure Service Bus a consumer receives a message, processes it and acknowledges it; the acknowledged message is deleted. In Kafka a topic is an append-only log split into partitions. A message is not deleted when it is read; it stays until the retention period expires. What the consumer manages is not the message but the read position in each partition, the offset. Every consumer group reading the same topic keeps its own offsets; Billing and Notification read the same messages without knowing about each other. Consumers within one group share the partitions: a partition is assigned to only one consumer in the group at a time.

This has three consequences in .NET code. First, there is no per-message ack; you commit an offset that says "I have read up to here". Second, retry and dead-lettering are not Kafka features but your code's job. Third, ordering is guaranteed only within a partition, and parallelism is bounded by the number of partitions. The conceptual side of ordering and partitioning is in the Messaging and Event-Driven Architecture series post "Message Ordering and Partitioning". Example environment: the bookstore.orders topic, 3 partitions, a single-broker Kafka 4.2.1 container (KRaft mode, no ZooKeeper).

Producer: Key, Acks and Idempotence

public sealed class OrderEventProducer : IDisposable
{
    private readonly IProducer<string, string> _producer;

    public OrderEventProducer(string bootstrapServers)
    {
        ProducerConfig config = new()
        {
            BootstrapServers = bootstrapServers,
            ClientId = "bookstore-order-api",
            Acks = Acks.All,
            EnableIdempotence = true,
            LingerMs = 5,
            // to pick the same partition as Java clients
            Partitioner = Partitioner.Murmur2Random
        };
        _producer = new ProducerBuilder<string, string>(config).Build();
    }

    public async Task<DeliveryResult<string, string>> PublishAsync(OrderPlaced order, CancellationToken ct)
    {
        Message<string, string> message = new()
        {
            // the same key always goes to the same partition: ordering per order
            Key = order.OrderId.ToString(),
            Value = JsonSerializer.Serialize(order, MessageJson.Options),
            Headers = new Headers
            {
                { "message-id", Encoding.UTF8.GetBytes(Guid.NewGuid().ToString()) },
                { "type", Encoding.UTF8.GetBytes(OrderPlaced.MessageType) }
            }
        };
        return await _producer.ProduceAsync("bookstore.orders", message, ct);
    }

    public void Dispose()
    {
        // do not close without sending the buffered messages
        _producer.Flush(TimeSpan.FromSeconds(10));
        _producer.Dispose();
    }
}

IProducer is a thread-safe and expensive object: behind it are librdkafka's own threads, connections and send buffer. It should live as a single instance (singleton) for the lifetime of the application. Delivery reports for six orders and two repeats of the same order:

key=c17ccad4 -> partition 0, offset 2, status Persisted
key=b183dc39 -> partition 2, offset 0, status Persisted
key=ad6a3f91 -> partition 1, offset 6, status Persisted
key=70097c09 -> partition 0, offset 3, status Persisted
key=061b6a13 -> partition 0, offset 4, status Persisted
key=e60d9c0c -> partition 1, offset 7, status Persisted
key=c17ccad4 -> partition 0, offset 5 (same key again)
key=c17ccad4 -> partition 0, offset 6 (same key again)

The key's hash decides the partition; the same key went to the same partition again, with increasing offsets. If you key all of an order's events by the order id, that order's events are read in order. Messages without a key are spread across partitions and have no ordering between them.

The reasoning behind the settings:

• Acks = Acks.All: the broker acknowledges once the message is written to all in-sync replicas. That is already librdkafka's default, but writing it out documents the intent.
• EnableIdempotence = true: the producer's own retries do not create duplicates or reordering in the partition. librdkafka's default is false. This solves duplicates on the producer side; it does not solve duplicates on the consumer side.
• LingerMs: how long to wait for a batch to fill before sending; librdkafka's default is 5 ms. Raising it at high volume increases throughput and latency.
• Partitioner = Murmur2Random: librdkafka's default partitioner is consistent_random, which hashes the key with CRC32 (Cyclic Redundancy Check); the Java client uses murmur2. If both .NET and Java services write to the same topic, the same key can go to two different partitions and per-order ordering silently breaks. According to the librdkafka documentation, Murmur2Random is functionally equivalent to the Java producer's default partitioner.

Flush and Dispose

ProduceAsync waits until the delivery report arrives; Produce puts the message in the buffer and returns immediately, reporting the result through a callback. At high volume Produce is much faster, but buffered messages have not been sent yet. The second question at the top, with LingerMs = 1000:

%4|1794104763.788|TERMINATE|rdkafka#producer-1| [thrd:app]: Producer terminating with 100 messages (100 bytes) still in queue or transit: use flush() to wait for outstanding message delivery
flush=False: delivered 0 of 100
Flush returned, still in queue: 0
flush=True: delivered 100 of 100

The producer disposed without calling Flush sent none of the 100 messages. librdkafka wrote the warning to stderr, but there was no exception. Flush(timeout) returns the number of messages still in the buffer; if it is not zero, you should log it. That is why flushing the producer before disposing it is mandatory during graceful shutdown.

Running the Consumer Inside a BackgroundService

public sealed class OrderPlacedConsumer(
    IConfiguration configuration, InvoiceService invoices, ILogger<OrderPlacedConsumer> logger) : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        ConsumerConfig config = new()
        {
            BootstrapServers = configuration["Kafka:BootstrapServers"],
            GroupId = "billing",
            ClientId = $"billing-{Environment.ProcessId}",
            AutoOffsetReset = AutoOffsetReset.Earliest,
            EnableAutoCommit = true,
            // we mark the offset ourselves once processing is done
            EnableAutoOffsetStore = false
        };

        using IConsumer<string, string> consumer = new ConsumerBuilder<string, string>(config)
            .SetPartitionsAssignedHandler((_, partitions) =>
                logger.LogInformation("Assigned: {Partitions}", string.Join(", ", partitions.Select(p => p.Partition.Value))))
            .SetPartitionsRevokedHandler((_, partitions) =>
                logger.LogInformation("Revoked: {Partitions}", string.Join(", ", partitions.Select(p => p.Partition.Value))))
            .SetErrorHandler((_, error) => logger.LogWarning("Kafka error: {Reason}", error.Reason))
            .Build();

        consumer.Subscribe("bookstore.orders");

        try
        {
            while (!stoppingToken.IsCancellationRequested)
            {
                // Consume is synchronous and blocking: it waits until a message arrives or the token is cancelled
                ConsumeResult<string, string> result = consumer.Consume(stoppingToken);

                OrderPlaced order = JsonSerializer.Deserialize<OrderPlaced>(result.Message.Value, MessageJson.Options)!;
                await invoices.CreateInvoiceAsync(order, stoppingToken);

                // only the offset of a processed message becomes eligible for commit
                consumer.StoreOffset(result);
                logger.LogInformation("Stored offset {Partition}@{Offset}",
                    result.Partition.Value, result.Offset.Value);
            }
        }
        catch (OperationCanceledException)
        {
            // normal shutdown
        }
        finally
        {
            // commits the stored offsets and leaves the group cleanly
            consumer.Close();
            logger.LogInformation("Consumer closed");
        }
    }
}

The RabbitMQ and Service Bus clients worked with an event that pushes messages. Confluent.Kafka's consumer uses a pull model, and Consume is a synchronous, blocking call; there is no async version. This loop keeps a thread busy all the time. In .NET 10 the whole of BackgroundService.ExecuteAsync runs on a background thread, so this loop does not prevent the host from starting. In .NET 8 and 9, however, the part of ExecuteAsync before the first await ran synchronously during startup; a Consume loop blocking on the first line could stop other hosted services and Kestrel from ever starting. That is why starting the loop separately with Task.Run is a common habit on older versions. The consumer instance is not thread-safe; call it from one thread at a time.

consumer.Close() matters: it commits the stored offsets and tells the group coordinator "I am leaving". The partitions of a consumer that dies without closing are not assigned to anyone until the broker notices its death through the session timeout (librdkafka's default is 45 seconds).

Offset Commits: The Auto Commit Trap and At-Least-Once

librdkafka's defaults: enable.auto.commit = true, auto.commit.interval.ms = 5000, enable.auto.offset.store = true. The last setting means the offset of every message Consume returns becomes eligible for commit, whether or not it was processed. The answer to the first question:

0.2s processing 2@0
0.9s processing 1@0
1.6s processing 1@1
2.3s processing 1@2
3.0s processing 1@3
3.7s crash while processing 1@3
exit=1

GROUP           TOPIC            PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
lab-autocommit  bookstore.orders 0          -               7               -
lab-autocommit  bookstore.orders 1          4               8               4
lab-autocommit  bookstore.orders 2          1               1               0

In Kafka the committed offset is "the next message to read". The committed value for partition 1 is 4: when the group restarts, it will never see message 3. The order being processed was skipped without any error record. That is at-most-once behaviour; the real Kafka form of the concept in the "Delivery Guarantees" post of the Messaging and Event-Driven Architecture series.

The fix is in the Billing code: EnableAutoOffsetStore = false and StoreOffset(result) after processing. Auto commit stays on, but it only commits offsets we have marked. The price is duplicates. We killed Billing with kill -9 while it was processing six new orders; the processing finished and the offsets were stored, but the 5-second commit interval had not elapsed:

      Assigned: 0, 1, 2
      Stored offset 0@7
      Stored offset 0@8
      Stored offset 1@8
      Stored offset 2@1
      Stored offset 2@2
      Stored offset 2@3

TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG
bookstore.orders 0 7 9 2
bookstore.orders 1 8 9 1
bookstore.orders 2 1 4 3

The committed offsets did not move at all. The restarted Billing's log wrote exactly the same six lines again: six invoices were issued twice. This is at-least-once: no loss, but duplicates. Shrinking the commit interval narrows the window but does not close it; a synchronous Commit after every message seriously reduces throughput. The right answer is to make the consumer idempotent; we will add that to Billing in the series' idempotent consumer post.

Consumer Groups and Rebalancing

While Billing's first instance was running we started a second one, then stopped the second one. The first instance's log (six lines in the middle shortened):

      Assigned: 0, 1, 2
      Stored offset 2@0
      Stored offset 1@0
      Stored offset 1@1
      Stored offset 1@2
      Revoked: 0, 1, 2
      Assigned: 0, 1
      Stored offset 0@0
      ...
      Stored offset 1@3
      Revoked: 0, 1
      Assigned: 0, 1, 2
      Stored offset 1@4

The second instance got Assigned: 2 and gave Revoked: 2 when it shut down. Two things to notice. First, there are no duplicates on partition 1: during the revoke, the offsets stored up to 1@2 were committed, and after reassignment reading continued from 1@3. Second, when the second instance joined, the first instance released not just partition 2 but all three. That is the "eager" behaviour of librdkafka's default range,roundrobin strategy: on every rebalance everyone releases everything and waits for redistribution. PartitionAssignmentStrategy = CooperativeSticky releases only the partitions that change hands.

During the experiment, the output of kafka-consumer-groups.sh --describe (consumer identity columns removed) shows consumer lag:

GROUP           TOPIC            PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
billing         bookstore.orders 0          3               7               4
billing         bookstore.orders 1          3               8               5
billing         bookstore.orders 2          1               1               0

Lag is the difference between the last written offset and the offset the group has committed, and it is the most important health metric of a Kafka consumer. If it keeps growing, the consumers cannot keep up. But adding instances has a ceiling: in a topic with 3 partitions, a fourth consumer gets no partition and sits idle.

With Kafka 4.0 the new consumer protocol defined in KIP (Kafka Improvement Proposal) 848 became generally available; in Confluent.Kafka it has been production-ready since 2.12, but the default is still classic. In this protocol partition assignment happens on the broker and rebalances are incremental. Turning it on is one line:

ConsumerConfig config = new()
{
    BootstrapServers = Bootstrap, GroupId = "billing-848", AutoOffsetReset = AutoOffsetReset.Earliest,
    GroupProtocol = GroupProtocol.Consumer
};

Assigned: 0, 1, 2
Consumed 23 messages with group.protocol=consumer
---
InvalidOperationException: `session.timeout.ms` is not supported for `group.protocol=consumer` or share consumer. It is defined broker side

GROUP                            TYPE
billing-848                      Consumer
billing                          Classic

In the new protocol some settings such as session.timeout.ms live on the broker rather than the client; if you carry the old config over, you get an exception when the consumer is created. Read Confluent's migration guide before moving an existing group to the new protocol.

Poison Messages and the Dead-Letter Topic

We wrote a single message containing invalid JSON (JavaScript Object Notation) to partition 2. The log of the consumer above:

info: Bookstore.Billing.Kafka.OrderPlacedConsumer[0]
      Revoked: 0, 1, 2
info: Bookstore.Billing.Kafka.OrderPlacedConsumer[0]
      Consumer closed
fail: Microsoft.Extensions.Hosting.Internal.Host[9]
      BackgroundService failed
      System.Text.Json.JsonException: Expected depth to be zero at the end of the JSON payload. There is an open JSON object or array that should be closed. Path: $.orderId | LineNumber: 0 | BytePositionInLine: 12.
crit: Microsoft.Extensions.Hosting.Internal.Host[10]
      The HostOptions.BackgroundServiceExceptionBehavior is configured to StopHost. A BackgroundService has thrown an unhandled exception, and the IHost instance is stopping. To avoid this behavior, configure this to Ignore; however the BackgroundService will not be restarted.

The host stopped and the offset did not move. Even if the container orchestrator restarts the process, it will read the same message from the same offset and die the same way. In RabbitMQ the nack, in Service Bus MaxDeliveryCount, broke this loop; Kafka has no breaker, and the rest of the partition stays locked behind this message. The process's exit code being 0 is a separate surprise: the orchestrator may take it as a "clean shutdown".

The fix is to write permanent failures to a dead-letter topic and move the offset forward:

catch (JsonException ex)
{
    // a poison message must not lock the partition: write it to the dead-letter topic and move on
    await deadLetters.ProduceAsync("bookstore.orders.billing.dlt", new Message<string, string>
    {
        Key = result.Message.Key,
        Value = result.Message.Value,
        Headers = new Headers
        {
            { "error", Encoding.UTF8.GetBytes(ex.Message) },
            { "source", Encoding.UTF8.GetBytes($"{result.Topic}/{result.Partition.Value}@{result.Offset.Value}") }
        }
    }, stoppingToken);
    consumer.StoreOffset(result);
    logger.LogWarning("Dead-lettered {Partition}@{Offset}: {Error}",
        result.Partition.Value, result.Offset.Value, ex.Message);
    continue;
}

warn: Bookstore.Billing.Kafka.OrderPlacedConsumer[0]
      Dead-lettered 2@4: Expected depth to be zero at the end of the JSON payload. There is an open JSON object or array that should be closed. Path: $.orderId | LineNumber: 0 | BytePositionInLine: 12.

Looking at the DLT (dead-letter topic) with kafka-console-consumer.sh, the message is there with the error and its source position:

error:Expected depth to be zero at the end of the JSON payload. There is an open JSON object or array that should be closed. Path: $.orderId | LineNumber: 0 | BytePositionInLine: 12.,source:bookstore.orders/2@4	bad-1	{"orderId":

For transient failures (the database is unreachable for a while), instead of writing to the DLT right away you need retry topics and delayed retries; we will build that in the series' retry post. The distinction here is simple: an error that retrying will not fix goes to the DLT, an error that may fix itself goes to retry.

Production Notes

Think about the partition count up front. The number of partitions sets the maximum parallelism within a group and can be increased later, but increasing it changes how keys map to partitions; old and new events of the same order can land in different partitions. Decreasing it is not possible.

Know max.poll.interval.ms. If this much time (librdkafka's default is 5 minutes) passes between two Consume calls, the consumer is kicked out of the group and its partitions are given to someone else. If the work for a single message takes long, either raise this value or hand the work off elsewhere.

Watch lag and alert on it. Per-partition lag is the earliest warning sign in Kafka. We will collect metrics with OpenTelemetry in the series' observability post.

Choose retention based on the longest outage of your consumers. If a consumer is down for 8 days and retention is 7 days, messages written on the first day are deleted before they are read.

Keep the producer a singleton, and one consumer per thread. If one process needs several consumers, run each in its own loop as a separate instance.

Common Pitfalls

Expecting at-least-once with the default auto offset store. As we saw, the message being processed can be skipped. EnableAutoOffsetStore = false and StoreOffset after processing.

Disposing without Flush. No error, no messages.

Thinking Consume is async. Running a blocking call in an infinite loop on the thread pool permanently occupies a thread; on .NET 8/9 it can even stall startup.

Having .NET and Java services write to the same topic with different partitioners. Ordering silently breaks.

Letting a poison message escape as an exception. The partition locks, the host stops, and every restart comes back to the same spot.

Trying to use Kafka as a queue. There is no per-message ack, no delayed redelivery and no "drain the queue"; if you need those, RabbitMQ or Service Bus is the better tool.

A Small Experiment

★☆☆ Run Billing as three instances, then start a fourth. See in the kafka-consumer-groups.sh --describe output that the fourth instance gets no partition.

★★☆ Repeat the rebalance experiment with PartitionAssignmentStrategy = PartitionAssignmentStrategy.CooperativeSticky. Which partitions does the first instance release when the second joins? Compare the log with the eager behaviour.

★★★ Add a retry topic to Billing: write a message that hit a transient error to bookstore.orders.billing.retry, carrying the attempt count in a header; a separate consumer processes this topic with a 30-second delay and sends the message to the DLT after the third attempt.

We have seen the .NET clients of three brokers, and in all of them we serialized the message body as JSON. But what happens when we add a new field to OrderPlaced, or rename a field? In the next post we cover message contracts, versioning, backward compatibility and Schema Registry in Kafka.