Raw Confluent.Kafka Client

By the end of this lesson you will be able to:

• read a ProducerConfig / ConsumerConfig and say which librdkafka default each omitted setting takes, and where it differs from the Java client;
• explain which work happens on your thread, on Confluent.Kafka's background poll task and inside librdkafka's native threads;
• build, use and shut down a producer and a consumer so that no queued message is silently dropped and the group is not left waiting;
• name the four ways an error reaches your code, and pick the right handler for each;
• choose object lifetimes: one shared producer, one consumer per consume loop.

Prerequisites: "Core Kafka Mental Model" (logs, offsets, acks, batching) and "Consumer Groups & Offsets" (what a group member and a committed offset are). Each major section ends with a short exercise — try it before opening the answer.

Why Start with the Raw Client

At some point, a Kafka-backed .NET service starts misbehaving in production. Messages are delayed. A consumer group rebalances unexpectedly. A producer silently drops messages during a restart. You open the issue tracker for MassTransit or the Confluent Schema Registry client, read through a dozen threads, and eventually land on a comment pointing at a configuration option of librdkafka (the native C library Confluent.Kafka wraps) — linger.ms, max.poll.interval.ms, acks — that someone had left at its default. The fix is one line. But unless you already understand what those options do and where they live in the stack, that comment is nearly incomprehensible.

This is the core argument for starting with the raw Confluent.Kafka client before reaching for any abstraction. The raw client is not a stepping stone you leave behind — it is the substrate that every higher-level library runs on top of, and the one whose behavior ultimately determines what your service does in production.

The Abstraction Stack Is Real, and It Has a Floor

When you use MassTransit with Kafka, or any of the growing number of .NET libraries that integrate with Kafka, you are working with wrappers around Confluent.Kafka. Those wrappers configure a producer or consumer on your behalf, set up serialization, manage consumer group lifecycles, and expose a friendlier interface. (The Confluent Schema Registry client is different: it is a REST client plus serializers that you plug into Confluent.Kafka's builders — it never builds a producer or consumer for you.) That friendliness is genuinely valuable — for day-to-day development work, you probably want those abstractions.

But abstractions do not eliminate the underlying behavior; they encapsulate it. When something goes wrong at the abstraction layer, the failure is almost always a consequence of what the underlying Confluent.Kafka client is doing — a delivery timeout because the message timeout is too low, a consumer that never commits because the abstraction chose manual offset management and the calling code doesn't realize it, or a rebalance storm because the poll interval is tuned for a different workload.

┌──────────────────────────────────────────┐
│         Your Application Code            │
├──────────────────────────────────────────┤
│  MassTransit and other integration       │
│  libraries built on Confluent.Kafka      │
├──────────────────────────────────────────┤
│         Confluent.Kafka (.NET)           │  ← this lesson
├──────────────────────────────────────────┤
│     librdkafka (native C library)        │
├──────────────────────────────────────────┤
│      Kafka Broker (TCP/SASL/TLS)         │
└──────────────────────────────────────────┘

When you understand the Confluent.Kafka layer — its configuration model, its threading assumptions, its delivery guarantees, its error surfaces — debugging the abstraction layers above it becomes tractable. The issue in MassTransit becomes "the consumer isn't committing offsets because the abstraction turned auto-commit off and commits only after a step my handler never reaches," not an inscrutable mystery.

💡 Mental Model: Think of Confluent.Kafka as the engine in a car. MassTransit and similar libraries are the dashboard and the automatic transmission — they make driving easier, but when the engine light comes on, you need to understand what the engine is actually doing.

Every Config Option Has a Measurable Effect

The Confluent.Kafka client exposes its configuration through plain C# objects — ProducerConfig, ConsumerConfig, and AdminClientConfig — that map almost directly to the underlying librdkafka configuration properties. The defaults are chosen for reasonable general-purpose behavior. But "reasonable general-purpose" is not the same as "correct for your workload."

⚠️ Read defaults from librdkafka's documentation, not the Java client's. This is one of the most common sources of confusion in .NET Kafka work: most Kafka material online documents the JVM client, and several defaults differ. Where this lesson quotes a default, it's librdkafka's.

Consider a producer instantiated with only a bootstrap server:

// A producer with only the minimum required config.
// Every option not explicitly set takes a default from librdkafka.
var config = new ProducerConfig
{
    BootstrapServers = "localhost:9092"
    // Not set: Acks, LingerMs, BatchSize, CompressionType,
    // MessageTimeoutMs, MessageMaxBytes, and many more.
};

using var producer = new ProducerBuilder<string, string>(config).Build();

This compiles and runs. But the omitted options control fundamental behaviors:

• acks (librdkafka default: all) determines how many broker replicas must acknowledge before the client considers a message delivered. Some teams intentionally set 1 or 0 for throughput-sensitive workloads where occasional loss is acceptable. A choice is being made either way. (The Java client historically defaulted to 1, which is one reason copied advice misleads here.)
• linger.ms (librdkafka default: 5) controls how long the client waits to accumulate messages into a batch. 0 minimizes latency at the cost of throughput; higher values improve batching efficiency. The Java client defaulted to 0 until Kafka 4.0 and uses 5 now, so older JVM material disagrees with both current clients.
• message.timeout.ms (librdkafka default: 300000, five minutes) is the ceiling on how long the client retries a delivery before giving up. librdkafka accepts delivery.timeout.ms as an alias for this setting — but note the Java client's delivery.timeout.ms defaults to two minutes, not five.
• enable.idempotence (librdkafka default: false) makes the broker discard duplicates created by the producer's own retries; the Java client has defaulted to true since Kafka 3.0. This is one of the defaults that still differs today; "Delivery Semantics & Schema Strategy" covers what it buys you.

🎯 Key Principle: Configuration ignorance is not neutrality. Every config option you don't set still takes a value — the default — and that default shapes the behavior of your system.

💡 Real-World Example: A team discovers their consumer is being repeatedly removed from its group and triggering rebalances under load. The root cause is max.poll.interval.ms — the maximum time librdkafka allows between your calls into Consume before it concludes the application is stuck and voluntarily leaves the group. Their message processing is slow enough to exceed this window. A developer who has never seen the setting in the raw client has no frame of reference for what it means when it appears in a MassTransit issue thread.

What This Lesson Covers — and What Comes Next

This lesson focuses on two things: understanding the architecture of Confluent.Kafka and getting a working producer and consumer off the ground. acks and batching were introduced in Core Kafka Mental Model; idempotence, transactions and schemas are covered in Delivery Semantics & Schema Strategy. Offset management and manual commits are covered in Consumer & Manual Commits. Both later lessons assume the foundation built here.

⚠️ Common Mistake: The impulse when starting with Kafka in .NET is to reach immediately for an integration package, add it to the DI container, and start producing and consuming through its API. This works, right up until it doesn't. When something goes wrong in the abstraction, the developer who skipped the raw client has to reverse-engineer both the abstraction's behavior and the underlying client's behavior simultaneously.

🤔 Did you know? The ProducerConfig and ConsumerConfig classes are generated from librdkafka's configuration documentation. Property names map almost one-to-one: linger.ms becomes LingerMs. Where librdkafka has two names for one setting (an alias), only one of them gets a property, and not always the same half of the pair: linger.ms is itself an alias of queue.buffering.max.ms and gets LingerMs, while delivery.timeout.ms gets none and is set through MessageTimeoutMs (message.timeout.ms). That near one-to-one naming makes cross-referencing the lower-level documentation practical whenever you need it — and it's why the librdkafka config reference, not the Java one, is the page to bookmark.

✍️ Exercise: predict the defaults

A .NET service and a Java service (Kafka 4.x client) each create a producer with nothing but the bootstrap servers set. For each setting, write down the value each producer runs with: acks, linger.ms, the delivery timeout, and idempotence.

How Confluent.Kafka Fits into the Kafka Ecosystem

When you install the Confluent.Kafka NuGet package, you are not getting a pure .NET implementation of the Kafka protocol. You are getting a thin managed wrapper around librdkafka, a battle-tested C library that handles the actual TCP connections, protocol framing, and internal queuing. That distinction explains a category of behaviors that can otherwise seem like inexplicable quirks: native memory that doesn't show up in .NET heap dumps, errors that originate in native code and reach you as error codes rather than managed stack traces, and platform-specific native binaries landing in your runtimes/ folder at publish time.

The librdkafka Layer

librdkafka powers a large share of non-JVM Kafka clients — Python's confluent-kafka, Go's confluent-kafka-go, and Confluent.Kafka for .NET all sit on top of it. The NuGet package, through its librdkafka.redist dependency, ships precompiled native binaries for each supported platform (linux-x64, linux-arm64, osx, win-x64, and so on). At runtime, the managed assembly uses P/Invoke to call into the appropriate one.

┌────────────────────────────────────────────┐
│            Your .NET Application           │
├────────────────────────────────────────────┤
│         Confluent.Kafka (managed C#)       │
│ IProducer<K,V> IConsumer<K,V> IAdminClient │
├────────────────────────────────────────────┤
│      librdkafka (native C library)         │
│   Protocol framing, TCP sockets, queuing   │
├────────────────────────────────────────────┤
│         Kafka Broker(s) over TCP           │
└────────────────────────────────────────────┘

The practical consequences of this layering:

• Native memory: The internal message queue and socket buffers live outside the managed heap. If your producer accumulates a backlog, the .NET GC won't see that pressure — librdkafka surfaces an error when its queue limit is hit, not an OutOfMemoryException.
• Platform binaries: Publishing a self-contained app or applying aggressive trimming requires ensuring the correct native binary is included. Misconfigured Docker images can strip it, producing a DllNotFoundException at startup rather than a compile error.
• Unmanaged error paths: Some conditions originate inside librdkafka and are marshaled back as KafkaException or delivered through callbacks you have to register. Which surface an error uses depends on the API you called — a distinction covered in detail under Mistake 2.

The Three Primary Types

The managed API surfaces three interfaces covering nearly every Kafka use case. Each is a distinct client with its own connection lifecycle — understanding this prevents a common class of resource and behavior bugs.

IProducer<TKey, TValue> publishes records to topics. It holds persistent broker connections, manages an internal in-memory queue of pending messages, and owns librdkafka's native background threads plus one managed poll task that serves delivery reports (the per-message success-or-failure notices). Built via ProducerBuilder<TKey, TValue>.

IConsumer<TKey, TValue> reads records and manages group coordination — heartbeating, partition assignment, rebalancing — while tracking offset state locally until committed. Like the producer, it owns its own connection lifecycle.

IAdminClient handles cluster-level operations: creating and deleting topics, describing configurations, listing consumer groups, inspecting cluster metadata. Less frequently instantiated in application code, but essential for tooling and integration tests that set up topic fixtures.

┌───────────────────────────────────────────────────────────────────────┐
│                          Application Process                          │
│                                                                       │
│  ┌───────────────────┐  ┌───────────────────┐  ┌───────────────────┐  │
│  │ IProducer<K,V>    │  │ IConsumer<K,V>    │  │ IAdminClient      │  │
│  │ - Internal queue  │  │ - Group member    │  │ (optional)        │  │
│  │ - Broker conns    │  │ - Offset tracking │  │ - Broker conns    │  │
│  └─────────┬─────────┘  └─────────┬─────────┘  └─────────┬─────────┘  │
│            │                      │                      │            │
└────────────┼──────────────────────┼──────────────────────┼────────────┘
             ▼                      ▼                      ▼
         Broker(s)              Broker(s)              Broker(s)

🎯 Key Principle: These three types are not cheap to construct. Build() starts the client's threads and immediately begins connecting to a bootstrap broker to fetch cluster metadata; connections to partition leaders and the group coordinator follow as soon as they are needed. Treat them as long-lived: build one producer (and admin client) and share it, build one consumer per consume loop, and dispose them explicitly at shutdown.

Serialization as a First-Class Concern

The generic type parameters on IProducer<TKey, TValue> and IConsumer<TKey, TValue> are not syntactic sugar. At construction time the builder resolves a concrete ISerializer<T> for the producer side and IDeserializer<T> for the consumer side. The library ships built-in implementations for a specific, short list: byte[], string (UTF-8), int, long, float, double, and Null (plus Ignore on the deserializer side, for a key or value you want to skip).

⚠️ For serializers, that list is exhaustive. There is no built-in serializer for Guid, DateTime, decimal, or any other common .NET type. Assuming there is compiles fine and fails at runtime: Build() throws because no default serializer exists for the type. Serialize a Guid as a string or supply your own ISerializer<Guid>.

// Producer with string keys and byte[] values.
// The serializer pair is resolved at build time, not per-message.
using var producer = new ProducerBuilder<string, byte[]>(
    new ProducerConfig { BootstrapServers = "localhost:9092" })
    .SetKeySerializer(Serializers.Utf8)        // built-in ISerializer<string>
    .SetValueSerializer(Serializers.ByteArray) // built-in ISerializer<byte[]>
    .Build();

For custom types, implement ISerializer<T> directly:

public sealed class OrderEventSerializer : ISerializer<OrderEvent>
{
    public byte[] Serialize(OrderEvent data, SerializationContext context)
    {
        // context carries topic name and whether this is a key or value,
        // useful if a single serializer needs topic-aware behavior.
        return JsonSerializer.SerializeToUtf8Bytes(data);
    }
}

using var producer = new ProducerBuilder<string, OrderEvent>(
    new ProducerConfig { BootstrapServers = "localhost:9092" })
    .SetKeySerializer(Serializers.Utf8)
    .SetValueSerializer(new OrderEventSerializer())
    .Build();

SerializationContext carries the topic name, a MessageComponentType (Key or Value) and the message headers. This matters if you later adopt the Schema Registry client, which uses the topic name to look up the right schema — the plumbing is already there at the raw level.

💡 Real-World Example: A common trajectory is to start with IProducer<string, byte[]> and serialize JSON by hand, then migrate to a typed IProducer<string, OrderEvent> with a custom ISerializer<T>. Because serialization is registered at construction rather than per-message, the serializer itself is one line on the builder. The rest of the change is mechanical and easy to find: the producer's type argument (in the builder and wherever the producer is injected), and call sites that built byte[] values now pass OrderEvent objects. There are no per-message serialization calls scattered through the code to hunt down.

The Internal Queue, Background Threads and Message Buffering

When you call Produce() or ProduceAsync(), your serializer runs on the calling thread, and then the message does not go to the broker immediately. librdkafka writes it into an internal in-memory queue, and its native background threads drain that queue, batch by topic-partition, and send over TCP.

Your Code
   │
   │  Produce() / ProduceAsync()
   ▼
┌─────────────────────────────────┐
│   librdkafka Internal Queue     │  ← native memory
│   [msg1][msg2][msg3]...         │
└──────────────┬──────────────────┘
               │  batched by linger.ms / batch.size
               ▼
┌─────────────────────────────────┐
│   Background I/O threads        │
│   (owned by librdkafka)         │
└──────────────┬──────────────────┘
               │  TCP
               ▼
          Kafka Broker
               │
               │  Acknowledgment (per acks setting)
               ▼
┌─────────────────────────────────┐
│   Confluent.Kafka poll task:    │
│   Delivery Report Callback      │
│   or ProduceAsync Task result   │
└─────────────────────────────────┘

Concrete implications:

• A successful Produce() call means the message entered the queue, not that it reached the broker. The delivery guarantee only holds once the delivery callback fires with no error or the ProduceAsync task completes without throwing.
• Disposing a producer without flushing risks losing messages still in the queue. Flagged here so the mental model is right from the start; the bootstrapping section returns to it.
• Two kinds of background thread are involved. librdkafka's own threads are native — you cannot await them or see them in TPL tooling. Results come back through Confluent.Kafka's managed poll task (a long-running task each producer starts), which calls librdkafka's poll and runs your delivery handlers and completes ProduceAsync tasks. That task — not your calling thread — is where delivery callbacks execute.

🤔 Did you know? librdkafka sends group heartbeats from its own background thread — and so does the Java client, since Kafka 0.10.1 (KIP-62) — so a slow handler does not miss heartbeats or trip session.timeout.ms. What it does trip is max.poll.interval.ms — the client-side limit on the gap between your Consume() calls. Exceed it and librdkafka logs a warning starting Application maximum poll interval (300000ms) exceeded (the number is your configured value), leaves the group, and your next Consume throws a ConsumeException (Local_MaxPollExceeded); a commit for the partitions it held fails because you no longer own them.

At this point the architecture resolves into a coherent picture: the NuGet package is a managed surface over a native library; the three primary types each own independent connections and threads; serialization is wired in at construction time; and wire I/O is always asynchronous with respect to your Produce() calls, mediated by an internal queue, librdkafka's threads and a managed poll task.

✍️ Exercise: where does it run, where does it live?

A service produces 50,000 small messages per second through one IProducer<string, string> using Produce(..., handler). Answer each:

1. The broker slows down and messages pile up in the producer until its queue is full (by default 100,000 messages, two seconds' worth at this rate). Will a managed heap dump show them? What happens when the queue limit is reached?
2. On which thread does the delivery handler run?
3. The handler throws an exception. Where does it surface?
4. A teammate changes the key type to Guid without adding a serializer. When does it fail?

Bootstrapping a Producer and Consumer: A Minimal Working Example

The gap between "I have Confluent.Kafka installed" and "I have a producer and consumer that work correctly" is smaller than it looks — but it contains several decisions that are easy to get wrong silently. A dropped message or a consumer that never receives anything won't throw; you simply won't see the data you expected.

Configuring the Clients

ProducerConfig and ConsumerConfig are thin typed wrappers over a dictionary of librdkafka settings. You populate them before building, and the builder validates at construction rather than at the first send or receive — front-loading configuration errors is the correct tradeoff.

For a producer, exactly one field is non-negotiable: BootstrapServers, the comma-separated host:port list used to discover the cluster. (After initial connection the client learns full topology from Kafka itself — this is just the entry point.)

For a consumer, two are required: BootstrapServers and GroupId, which names the group this instance joins. Group identity governs partition assignment, offset tracking, and rebalance behavior. Leaving GroupId unset throws at build time — a consumer without a group is a malformed consumer.

// Minimal producer configuration
var producerConfig = new ProducerConfig
{
    BootstrapServers = "localhost:9092"
    // All other settings take librdkafka defaults. Delivery guarantees
    // (idempotence, transactions) are covered in "Delivery Semantics &
    // Schema Strategy".
};

// Minimal consumer configuration
var consumerConfig = new ConsumerConfig
{
    BootstrapServers = "localhost:9092",
    GroupId = "my-consumer-group",

    // Applies when this group has no committed offset for a partition, or
    // the committed offset is out of range (retention already deleted it).
    // librdkafka's default is Latest, which silently skips everything already
    // in the log — set it explicitly rather than inheriting that.
    AutoOffsetReset = AutoOffsetReset.Earliest
};

⚠️ Common Mistake: Treating AutoOffsetReset as a server-side setting you can leave to the cluster. It is purely client-side, it defaults to Latest, and it applies only when there is no usable committed offset — none yet, or one that points at data retention has already deleted. A brand-new group pointed at a long-running topic with the default in place reads nothing that was produced before it started — the classic "why did I miss messages?" bug.

One other default worth knowing before you write a consumer loop: EnableAutoCommit is true, and its companion EnableAutoOffsetStore is also true, which together mean offsets can advance before your handler finishes. The consequences and the two correct configurations are the subject of Consumer & Manual Commits; the examples here leave the defaults in place because they only print messages.

Building the Clients

ProducerBuilder<TKey, TValue> and ConsumerBuilder<TKey, TValue> are the public construction path (plus DependentProducerBuilder, for a producer that shares another client's handle). The concrete implementations are internal — you cannot new them directly. The builder lets the library enforce invariants and configure the underlying librdkafka handle before handing you the object.

using IProducer<string, string> producer =
    new ProducerBuilder<string, string>(producerConfig).Build();

using IConsumer<string, string> consumer =
    new ConsumerBuilder<string, string>(consumerConfig).Build();

Both implement IDisposable, and Dispose does real work — tearing down the librdkafka handle and closing TCP connections. On the producer side, flush behavior at dispose deserves attention.

Disposing Correctly: The Producer Flush Problem

librdkafka buffers messages internally and sends them in batches on a background thread. That creates a hazard at shutdown. Disposal does not drain the queue. Dispose stops the producer's poll task and destroys the librdkafka handle, and librdkafka's documented termination sequence is flush first, then destroy: messages still queued or in flight at destroy time are dropped, with no exception and no delivery report.

The reliable pattern is to flush with a bound you chose, and to check what it returns:

// Flush returns the number of messages STILL unconfirmed (queued or in
// flight) when the timeout expired. Anything above zero may be lost, or
// may already be written; you won't find out which.
var stillQueued = producer.Flush(TimeSpan.FromSeconds(10));
if (stillQueued > 0)
    logger.LogError("{Count} messages were never confirmed", stillQueued);

producer.Dispose();

// With a using block, flush explicitly before the scope ends:
using (var p = new ProducerBuilder<string, string>(producerConfig).Build())
{
    // ... produce messages ...
    var left = p.Flush(TimeSpan.FromSeconds(10)); // BEFORE the using block exits
    if (left > 0)
        logger.LogError("{Count} messages will be dropped on dispose", left);
} // Dispose called here

⚠️ Common Mistake: Relying on Dispose alone to drain the buffer, and never looking at what Flush returns. At shutdown, a non-zero return is your code's only sign that Dispose is about to drop messages — Dispose raises no exception for them. A message that fails during the flush is reported only through its delivery report or ProduceAsync task.

Sending a Message: ProduceAsync vs. Produce with a Delivery Handler

Sending uses either ProduceAsync (awaitable, returns DeliveryResult<TKey, TValue>) or Produce with a delivery handler callback. Both are legitimate; the choice depends on throughput requirements.

try
{
    var deliveryResult = await producer.ProduceAsync(
        topic: "my-topic",
        message: new Message<string, string>
        {
            Key = "order-id-123",
            Value = "{\"amount\": 42.00}"
        });

    Console.WriteLine(
        $"Delivered to partition {deliveryResult.Partition}, "
        + $"offset {deliveryResult.Offset}, status {deliveryResult.Status}");
}
catch (ProduceException<string, string> ex)
{
    Console.Error.WriteLine($"Delivery failed: {ex.Error.Reason}");
}

The tradeoff: awaiting each call before sending the next introduces per-message latency, because you wait for a broker acknowledgment before moving on. Awaiting does not block a thread — the method is suspended and the thread returns to the pool — but the next message is not produced until the acknowledgment arrives, so for high-throughput scenarios that round trip dominates. (Starting many ProduceAsync calls and awaiting them together avoids this.)

The Produce overload taking an Action<DeliveryReport<TKey, TValue>> returns immediately; Confluent.Kafka invokes your callback on the producer's background poll task when the broker responds.

producer.Produce(
    topic: "my-topic",
    message: new Message<string, string>
    {
        Key = "order-id-456",
        Value = "{\"amount\": 17.50}"
    },
    deliveryHandler: report =>
    {
        // DeliveryReport (not DeliveryResult) is what carries Error.
        if (report.Error.IsError)
            Console.Error.WriteLine($"Delivery failed: {report.Error.Reason}");
        else
            Console.WriteLine($"Delivered to offset {report.Offset}");
    });

// Messages are still in the internal queue here — flush before disposing.
producer.Flush(TimeSpan.FromSeconds(10));

⚠️ Common Mistake: Assuming the callback fires before Produce returns. It does not, and it runs on the producer's background poll task, not your thread, so any state it closes over must be thread-safe and the handler itself must be fast and must not throw.

Subscribing and Consuming

There are two ways to get partitions: Subscribe and Assign. They are not interchangeable.

Subscribe registers the consumer with the group coordinator. The group divides the partitions among all consumers sharing the GroupId (under the classic protocol, librdkafka's default, one member computes the split and the coordinator hands it out) and rebalances as members join and leave. This is the standard path.

Assign skips group assignment and pins the consumer to specific TopicPartitions. Offset commits — automatic ones included, since EnableAutoCommit defaults to true — still go to the group coordinator under the GroupId (Confluent.Kafka requires one on every consumer anyway), but nobody manages who reads what: two instances assigning the same partition both read it and overwrite each other's commits. It suits narrow cases — replaying a partition from a known offset while debugging — and is not a substitute for Subscribe.

After subscribing, the loop calls Consume with a cancellation token. Each call drives the client's internal machinery and delivers the next available message.

consumer.Subscribe("my-topic");

// In a BackgroundService this is the framework's stoppingToken.
using var cts = new CancellationTokenSource();
Console.CancelKeyPress += (_, e) => { e.Cancel = true; cts.Cancel(); };

try
{
    while (!cts.IsCancellationRequested)
    {
        // The CancellationToken overload blocks until a message arrives
        // or the token is cancelled — it never returns null. Only the
        // timeout overloads (TimeSpan / int milliseconds) return null.
        var result = consumer.Consume(cts.Token);

        Console.WriteLine(
            $"Received: key={result.Message.Key} "
            + $"value={result.Message.Value} "
            + $"partition={result.Partition} "
            + $"offset={result.Offset}");
    }
}
catch (OperationCanceledException)
{
    // Normal shutdown path when the token is cancelled.
}
finally
{
    // Close() leaves the group (and commits final offsets if auto-commit
    // is on), then disposes the consumer.
    consumer.Close();
}

Three things deserve attention.

consumer.Close() in the finally tells the group coordinator this member is leaving, triggering an immediate rebalance rather than waiting out session.timeout.ms. (Under the classic group protocol that is a LeaveGroup request; the newer KIP-848 protocol signals it differently, with the same effect, and there the timeout is the broker's group.consumer.session.timeout.ms rather than the client's session.timeout.ms.) Skip it and other group members wait out the full timeout before picking up those partitions.

The CancellationToken passed to Consume is the standard way to stop the loop from outside; the overload throws OperationCanceledException when cancelled — within about 100 ms, since it polls librdkafka in short slices (dotnet.cancellation.delay.max.ms) — which is the idiomatic .NET exit path. Note that this exception is not a ConsumeException, so a catch that only handles the latter lets it escape.

And the null check you may have seen elsewhere is absent deliberately — Consume(CancellationToken) never returns null. Writing if (result is null) continue; after it implies a case that cannot occur.

✍️ Exercise: order the shutdown

A service owns one consumer loop (in a hosted service) and one shared producer. On SIGTERM, put these steps in the right order, and say what goes wrong if producer.Dispose() runs first.

• producer.Flush(timeout) and check the returned count
• the host cancels stoppingToken
• consumer.Close()
• Consume throws OperationCanceledException and the loop exits
• producer.Dispose()

Common Mistakes When Using the Raw Client

Working directly with Confluent.Kafka rewards attention to detail. The library gives you precise control, but that precision cuts both ways: persistent broker connections, internal threading, and non-exception error paths that higher-level abstractions handle quietly are now your responsibility.

Mistake 1: Creating a New IProducer Per Message

The most immediately costly mistake is treating IProducer<TKey, TValue> as a lightweight, stateless utility — constructing one, calling Produce, disposing it before the next message.

Each instance maintains its own broker connections and an internal message queue in native memory. Every Build() starts librdkafka's threads and Confluent.Kafka's poll task, allocates the buffer, and begins connecting to a bootstrap broker for metadata; the first produce then connects to the partition leader. Disposing tears all of it down — and drops whatever was still queued.

Per-message producer (wrong)
──────────────────────────────────────────────────────────────
  Message 1  →  [Build] [TCP handshake] [Produce] [Dispose]
  Message 2  →  [Build] [TCP handshake] [Produce] [Dispose]
                   ↑           ↑
                   │           └─ connection setup and round trips, every message
                   └─ new threads and native queue, every message

Shared producer (correct)
──────────────────────────────────────────────────────────────
  [Build once] [TCP handshake once]
  Message 1  →  [Produce → internal queue]
  Message 2  →  [Produce → internal queue]  →  [batch to broker]
  [Flush on shutdown] [Dispose]

Beyond throughput there's a resource leak dimension. A per-message or per-request producer that is ever not disposed is never collected: its running poll task keeps it reachable, and its queue, threads and sockets are native resources the GC cannot see. Production systems can exhaust file descriptors or native heap long before the managed heap shows pressure.

✅ Instantiate one IProducer per application (or per logical pipeline), register it as a singleton, and dispose it explicitly on shutdown after flushing.

Mistake 2: Not Knowing Where Errors Actually Surface

Kafka errors reach your code through four different mechanisms, and which one applies depends entirely on the API you called. Code that wraps everything in a single try/catch catches perhaps half of them.

⚠️ The API detail that trips people up: DeliveryResult<K,V> — what ProduceAsync returns — has no Error property. Neither does ConsumeResult<K,V>. Only DeliveryReport<K,V> (the callback type, which derives from DeliveryResult) carries Error. Code written as if (result.Error.IsError) after an await producer.ProduceAsync(...) does not compile, and the instinct behind it — that failures come back as a flag on the result — is the wrong model. ProduceAsync and Consume throw.

What DeliveryResult does give you is Status, a PersistenceStatus of Persisted, NotPersisted, or PossiblyPersisted. A result that comes back normally says Persisted (with acks=0, where no acknowledgment exists, it can only say PossiblyPersisted). With acks of 1 or all, the ambiguous case — a request that was sent but never confirmed — is a failure, so it arrives on the exception as ex.DeliveryResult.Status:

try
{
    await producer.ProduceAsync("orders", new Message<string, string>
    {
        Key = order.Id,
        Value = JsonSerializer.Serialize(order)
    });
}
catch (ProduceException<string, string> ex)
    when (ex.DeliveryResult.Status == PersistenceStatus.PossiblyPersisted)
{
    // Sent but never confirmed: it may already be in the log,
    // so a retry can write it twice.
    logger.LogWarning(ex,
        "Order {OrderId} may or may not have been written — treat as ambiguous",
        order.Id);
    throw;
}
catch (ProduceException<string, string> ex)
{
    logger.LogError(ex,
        "Delivery failed for order {OrderId}: [{Code}] {Reason} (status {Status})",
        order.Id, ex.Error.Code, ex.Error.Reason, ex.DeliveryResult.Status);
    throw;
}

On the consumer side, deserialization failures and protocol errors arrive as ConsumeException, which carries the raw record so you can route it somewhere:

try
{
    var result = consumer.Consume(cancellationToken);
    ProcessMessage(result.Message);
}
catch (ConsumeException ex)
{
    logger.LogWarning(
        "Consume error at [{Topic}/{Partition}@{Offset}]: [{Code}] {Reason}",
        ex.ConsumerRecord?.Topic, ex.ConsumerRecord?.Partition,
        ex.ConsumerRecord?.Offset, ex.Error.Code, ex.Error.Reason);
    // The client has already moved past this record: route the raw bytes in
    // ex.ConsumerRecord to a dead-letter topic now, or it is silently skipped.
}

The surface most teams never register at all is the fourth row: client-level errors that aren't attached to any single message. A broker becoming unreachable, an authentication failure, a DNS resolution problem — none of these correspond to a Produce or Consume call, so none of them throw anywhere your code is looking. They reach your code only through a handler you opt into on the builder:

using var producer = new ProducerBuilder<string, string>(config)
    .SetErrorHandler((_, error) =>
    {
        // Not tied to any one message. Without this handler librdkafka only
        // logs them (to stderr, or to your log handler) — nothing alerts.
        if (error.IsFatal)
            logger.LogCritical("Fatal Kafka error [{Code}]: {Reason}",
                error.Code, error.Reason);
        else
            logger.LogWarning("Kafka client error [{Code}]: {Reason}",
                error.Code, error.Reason);
    })
    .SetLogHandler((_, log) => logger.Log(
        (LogLevel)log.LevelAs(LogLevelType.MicrosoftExtensionsLogging),
        "[librdkafka] {Facility}: {Message}", log.Facility, log.Message))
    .Build();

💡 Pro Tip: Error.Code is your primary diagnostic. ErrorCode.Local_AllBrokersDown in the error handler tells you connectivity is gone even though no produce call has failed yet. ErrorCode.Local_QueueFull from Produce means the internal queue is saturated — a backpressure signal worth alerting on, and one that appears as a thrown ProduceException, not a silent drop.

One more note on ConsumeResult: it exposes IsPartitionEOF, which is true when you've reached the end of a partition. That flag is only ever set if you opt in with EnablePartitionEof = true — the default is false, so a check for it in default configuration is dead code.

Mistake 3: Calling Consume in an Endless Loop Without a Cancellation Token

Consume() with no cancellation token blocks indefinitely until a message is available or a consume error (not only a fatal one) makes it throw. The problem surfaces at deployment: when the orchestrator sends SIGTERM and expects the app to drain and exit within a grace period, a loop blocked on Consume() with no cancellation path never exits on its own. The process ends with the loop still blocked — the .NET host stops waiting after its shutdown timeout (30 s by default), or the orchestrator escalates to SIGKILL after its grace period — and Close() never runs.

// ⚠️ WRONG: blocks indefinitely, no way to stop gracefully
while (true)
{
    var result = consumer.Consume(); // infinite block
    Process(result);
}

// ✅ CORRECT: pass the application's cancellation token
try
{
    while (!stoppingToken.IsCancellationRequested)
    {
        var result = consumer.Consume(stoppingToken);
        Process(result);
    }
}
catch (OperationCanceledException)
{
    // expected on shutdown
}
finally
{
    consumer.Close();
}

Graceful shutdown sequence
──────────────────────────────────────────────────────────────
  Orchestrator       Application            Kafka Broker
      │                   │                       │
      │──── SIGTERM ─────▶│                       │
      │                   │ CancellationToken     │
      │                   │  .Cancel()            │
      │                   │                       │
      │                   │ Consume() throws      │
      │                   │ OperationCanceled     │
      │                   │                       │
      │                   │ consumer.Close():     │
      │                   │──── leave group ─────▶│
      │                   │                       │ Reassigns partitions
      │                   │◀─── response ─────────│  immediately
      │                   │ Close() also disposes │
      │                   │                       │
      │◀──── Process exit─│                       │

💡 Real-World Example: In a hosted BackgroundService, the stoppingToken passed to ExecuteAsync is exactly the token to use. When the host begins shutdown it cancels that token, unblocking Consume and letting ExecuteAsync return cleanly. On .NET 9 and earlier, the synchronous part of ExecuteAsync runs during host startup, so a blocking Consume loop placed directly inside it stalls every service registered after it — wrap the loop in Task.Run (or a dedicated long-running task). .NET 10 changed this: BackgroundService now runs all of ExecuteAsync on a background thread, so startup is no longer blocked, but the loop still occupies that thread for the life of the service.

Mistake 4: Sharing One IConsumer Across Threads

The two clients are meant to be used differently. Many threads calling Produce / ProduceAsync on one shared producer is the intended usage — each call is an independent enqueue. A consumer is not built to be shared: design for one consume loop driving each IConsumer.

The reason is not that librdkafka's C API would corrupt itself — librdkafka documents its API as thread-safe. It is the consume loop's semantics. Consume hands out one ordered stream per partition; rebalance and error handlers run inside Consume, on whichever thread happens to be calling it; a handler exception captured during one thread's call is rethrown to the next caller; and with the default offset auto-store, every record handed out is marked as done. Several threads pulling from one consumer therefore process a partition's records concurrently and out of order, and let stored offsets run ahead of work that hasn't finished — so a crash loses messages.

// ⚠️ WRONG: sharing one consumer across threads
for (int i = 0; i < workerCount; i++)
{
    Task.Run(() =>
    {
        while (true)
        {
            var result = consumer.Consume(stoppingToken); // ← records of one partition
                                                          //   now race each other
            Process(result);
        }
    });
}

// ✅ Option A: one consumer per thread, all sharing a GroupId.
// Note the ceiling: if workerCount exceeds the topic's partition count,
// the surplus consumers join the group and sit idle holding no partitions.
for (int i = 0; i < workerCount; i++)
{
    var threadConsumer = new ConsumerBuilder<string, string>(config).Build();
    threadConsumer.Subscribe("orders");

    Task.Run(() =>
    {
        try
        {
            while (!stoppingToken.IsCancellationRequested)
            {
                var result = threadConsumer.Consume(stoppingToken);
                Process(result);
            }
        }
        catch (OperationCanceledException) { }
        finally
        {
            threadConsumer.Close();
            threadConsumer.Dispose();
        }
    });
}

using System.Threading.Channels;

// ✅ Option B: single consume loop feeds a channel; parallelism downstream.
var channel = Channel.CreateBounded<ConsumeResult<string, string>>(capacity: 256);

_ = Task.Run(async () =>
{
    try
    {
        while (!stoppingToken.IsCancellationRequested)
        {
            var result = consumer.Consume(stoppingToken);
            await channel.Writer.WriteAsync(result, stoppingToken);
        }
    }
    catch (OperationCanceledException) { }
    finally
    {
        channel.Writer.Complete();
        consumer.Close();
        consumer.Dispose();
    }
});

await Parallel.ForEachAsync(
    channel.Reader.ReadAllAsync(stoppingToken),
    new ParallelOptions { MaxDegreeOfParallelism = workerCount, CancellationToken = stoppingToken },
    async (result, ct) => await ProcessAsync(result));

⚠️ Option B keeps a single consume loop on the consumer, so its calls never overlap, and creates two new problems. The poll loop keeps consuming while workers are still busy, so whatever advances the offset — auto-commit by default — runs ahead of actual processing, and a crash loses everything in flight. And several workers can process records from the same partition concurrently, discarding the per-partition ordering that Kafka's whole assignment model exists to protect. Making it correct requires tracking completion per partition and only committing the highest contiguously finished offset. Use the simple form only where losing in-flight records on a crash, and processing out of order, are both genuinely acceptable; Consumer & Manual Commits develops the correct version.

✍️ Exercise: spot the bugs

This hosted service "works on my machine". Find at least five problems before opening the answer.

public sealed class AuditForwarder : BackgroundService
{
    protected override Task ExecuteAsync(CancellationToken stoppingToken)
    {
        using var consumer = new ConsumerBuilder<string, string>(new ConsumerConfig
        {
            BootstrapServers = "kafka:9092",
            GroupId = "audit-forwarder"
        }).Build();
        consumer.Subscribe("orders");

        while (true)
        {
            try
            {
                var result = consumer.Consume(stoppingToken);
                if (result is null) continue;

                using var producer = new ProducerBuilder<string, string>(
                    new ProducerConfig { BootstrapServers = "kafka:9092" }).Build();
                producer.Produce("orders-audit", new Message<string, string>
                {
                    Key = result.Message.Key,
                    Value = result.Message.Value
                });
            }
            catch (Exception ex)
            {
                Console.WriteLine(ex.Message);
            }
        }
    }
}

Key Takeaways and What Comes Next

The Architecture Is the Explanation

Confluent.Kafka is a managed wrapper around librdkafka. When you see behavior that feels surprising from a .NET perspective — native memory pressure under high throughput, errors that arrive as native error codes, background threads that persist when the client looks idle — the explanation is almost always in librdkafka's architecture.

It also explains why disposal matters more than for purely managed clients. The background threads and internal queue are native resources, disposal drops whatever is still queued, and a Flush you control with a timeout you chose tells you, at shutdown, how many messages were still unconfirmed when you gave up. During normal operation each message's fate is told by its delivery report or ProduceAsync task; the messages Flush still counts when you give up get neither, because Dispose drops them without a report.

Long-Lived, Stateful Objects — Not Request-Scoped

Both client types own broker connections, internal queues, and background threads. Treating them as request-scoped multiplies construction cost with every operation and, for producers, drops every message still queued when each scope disposes its instance.

💡 Mental Model: IProducer is like HttpClient — one shared instance for the application, called from anywhere. IConsumer is long-lived too, but it is not shared: it belongs to one consume loop, usually inside a hosted service.

// ✅ Register as a singleton in your DI container
public static IServiceCollection AddKafkaProducer(
    this IServiceCollection services,
    IConfiguration configuration)
{
    services.AddSingleton<IProducer<string, string>>(sp =>
    {
        var config = new ProducerConfig
        {
            BootstrapServers = configuration["Kafka:BootstrapServers"]
        };
        return new ProducerBuilder<string, string>(config)
            .SetErrorHandler((_, e) => /* log client-level errors */ { })
            .Build();
    });

    // A hosted service whose StopAsync calls Flush(timeout) on the singleton,
    // so draining happens once at shutdown rather than per request.
    // Register it before the consumer's hosted service: the host stops
    // services in reverse order, so the flush runs after the loop ends.
    services.AddHostedService<KafkaShutdownService>();

    return services;
}

IConsumer is also long-lived, with an extra consequence if you get it wrong: each new instance is a new group member from the broker's perspective, so creating one per message continuously rebalances the group.

Serialization, Errors, and Cancellation Are Your Responsibility

Serialization: ISerializer<T> / IDeserializer<T> must be registered for any type outside the built-in list (byte[], string, int, long, float, double, Null, plus Ignore for deserializers). Guid is not on that list.

Error surfaces: Four distinct mechanisms. Two are exceptions from the call you made (ProduceAsync, Consume; Produce also throws for immediate failures such as a full queue); the delivery-report callback and the client-level error handler reach you on other threads, after the call returned. Registering SetErrorHandler is the difference between seeing a broker outage and watching throughput mysteriously drop.

Cancellation: The Consume(CancellationToken) overload is the correct shutdown path, and its OperationCanceledException is not a ConsumeException — catch both, separately.

What Changes in Your Mental Model

What Comes Next

Delivery Semantics & Schema Strategy takes the ProducerConfig populated here with only BootstrapServers and covers the settings that decide delivery guarantees — idempotence, max.in.flight.requests.per.connection, transactions — plus schema evolution. It assumes you understand the internal queue and background threads, which you now do. Production Patterns & Observability then covers shutdown, lag and error routing in a running service.

Consumer & Manual Commits covers offset management: EnableAutoCommit versus manual StoreOffset/Commit, what happens to uncommitted offsets during a rebalance, and how to avoid the duplicate-processing and message-loss modes that offset errors produce — including the correct version of the worker hand-off pattern from Mistake 4. It assumes you understand Subscribe versus Assign and why a consumer is a stateful group member.

✍️ Exercise (harder): forward without losing anything

Design the consume-and-forward service from the "spot the bugs" exercise properly. Requirements: every record read from orders must end up in orders-audit, even across crashes and deploys; duplicates in orders-audit are acceptable. Say (a) the lifetime of each client, (b) when a record's offset may be marked as done, and (c) the shutdown order.

🤔 Did you know? The consumer's commit logic talks to the group coordinator on the broker, not just local state — Commit() sends a network request rather than updating a local bookmark. That's why commit timing matters as much as it does, and why Consumer & Manual Commits warrants its own lesson.