Raw Confluent.Kafka Client

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 librdkafka configuration option — 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 the Confluent Schema Registry client, 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. 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 / Schema Registry Client /  │
│  Other Higher-Level Abstractions         │
├──────────────────────────────────────────┤
│         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 is calling Consume without a cancellation token and then exiting before the commit happens," 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 the single most reliable source 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 default is 0.
• 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.

🎯 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. Producer configuration — delivery guarantee knobs, batching, compression — is covered in the dedicated Producer Configuration lesson. Offset management, consumer group coordination, and manual commits are covered in Consumer & Manual Commits. Both 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. That symmetry 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.

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, exception stack traces that bottom out in unmanaged code, 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 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>  IAdmin   │
├────────────────────────────────────────────┤
│      librdkafka (native C library)         │
│   Protocol framing, TCP pooling, 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 a persistent TCP connection pool, manages an internal in-memory queue of pending messages, and owns a background I/O thread inherited from librdkafka. 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>          │     │
│  │  - Internal queue    │  │  - Group coordinator     │     │
│  │  - I/O thread        │  │  - Offset tracking       │     │
│  │  - Connection pool   │  │  - Connection pool       │     │
│  └──────────┬───────────┘  └──────────────┬───────────┘     │
│             │                             │                 │
│  ┌──────────┴─────────────────────────────┴───────────┐     │
│  │              IAdminClient (optional)               │     │
│  │  - Separate connection lifecycle                   │     │
│  └────────────────────────────────────────────────────┘     │
└─────────────────────────────────────────────────────────────┘
         │                        │
         ▼                        ▼
   Kafka Broker               Kafka Broker

🎯 Key Principle: These three types are not cheap to construct. Each establishes real broker connections when first used. Treat them as long-lived, application-scoped objects — instantiate once, reuse throughout, dispose 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.

⚠️ That list is exhaustive. There is no built-in serializer for Guid, DateTime, decimal, or any other common .NET type — a reasonable assumption that produces a build-time exception rather than a compile error. 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 and a MessageComponentType (Key or Value). This matters if you later adopt the Schema Registry client, which uses the topic name to resolve the correct schema subject — 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 custom ISerializer<T> backed by a binary format later. Because serialization is registered at construction rather than per-message, that migration is a one-line change in the builder.

The Internal I/O Thread and Message Buffering

When you call Produce() or ProduceAsync(), the message does not go to the broker immediately. librdkafka writes it into an internal in-memory queue, and a background I/O thread drains that queue, batches by topic-partition, and sends over TCP.

Your Code
   │
   │  Produce() / ProduceAsync()
   ▼
┌─────────────────────────────────┐
│   librdkafka Internal Queue     │  ← native memory
│   [msg1][msg2][msg3]...         │
└──────────────┬──────────────────┘
               │  batched by linger.ms / batch.size
               ▼
┌─────────────────────────────────┐
│   Background I/O Thread         │
│   (owned by librdkafka)         │
└──────────────┬──────────────────┘
               │  TCP
               ▼
          Kafka Broker
               │
               │  Acknowledgment (per acks setting)
               ▼
┌─────────────────────────────────┐
│   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 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.
• The background thread is not observable from managed code. You cannot await it or see it in TPL tooling. It surfaces results exclusively through callbacks and returned Task values.

🤔 Did you know? librdkafka runs group heartbeats on its own background thread, which is a meaningful difference from the Java client. A blocked handler does not immediately fail the session timeout the way it would on the JVM. 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 Application maximum poll interval exceeded, leaves the group, and your next commit fails because you no longer own the partition.

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 and a background thread.

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 plain C# POCOs. 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. Producer Configuration
    // (the follow-on lesson) covers batching, compression, and delivery
    // guarantees in depth.
};

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

    // Applies ONLY when this group has no committed offset for a partition.
    // 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 no committed offset exists. 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 only public construction path. 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 attempt to drain the queue, but the timeout behavior it applies is a library-version detail rather than something you control, and a drain against an unreachable broker can block your shutdown path until the orchestrator escalates to SIGKILL — losing the messages anyway.

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

// Flush returns the number of messages STILL in the queue when the
// timeout expired. Anything above zero is data you are about to lose.
var stillQueued = producer.Flush(TimeSpan.FromSeconds(10));
if (stillQueued > 0)
    logger.LogError("{Count} messages were never delivered", stillQueued);

producer.Dispose();

// With a using block, flush explicitly before the scope ends:
using (var p = new ProducerBuilder<string, string>(producerConfig).Build())
{
    // ... produce messages ...
    p.Flush(TimeSpan.FromSeconds(10)); // BEFORE the using block exits
} // Dispose called here

⚠️ Common Mistake: Relying on Dispose alone to drain the buffer, and never looking at what Flush returns. The return value is the only signal you get that messages didn't make it — there is no exception and no log entry.

Sending a Message: ProduceAsync vs. Fire-and-Forget

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 introduces per-message latency, because you wait for a broker acknowledgment before moving on. For high-throughput scenarios that round trip dominates.

The Produce overload taking an Action<DeliveryReport<TKey, TValue>> is non-blocking. It returns immediately; librdkafka invokes your callback on its internal thread 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 a librdkafka thread, so any state it closes over must be thread-safe and the handler itself must be fast and must not throw.

	ProduceAsync	Produce + Callback
Blocking?	Awaitable (async)	Non-blocking
Throughput	Lower per-message	Higher per-message
Error handling	catch (ProduceException<K,V>)	Check report.Error in callback
Best for	Simplicity, low volume	High-throughput pipelines

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 coordinator distributes partitions across all consumers sharing the GroupId and triggers rebalances as members join and leave. This is the standard path.

Assign bypasses the coordinator entirely and pins the consumer to specific TopicPartitions. Offset commits still work, but assignment is no longer managed. 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
        // Consume(TimeSpan) overload returns null, on timeout.
        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() commits final offsets and notifies the group coordinator
    // before Dispose tears down the connection.
    consumer.Close();
}

Three things deserve attention.

consumer.Close() in the finally sends a LeaveGroup request, triggering an immediate rebalance rather than waiting out 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 only clean way to stop the loop from outside; the overload throws OperationCanceledException when cancelled, 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.

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: connection pooling, 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 TCP connection pool and an internal message queue in native memory. Every Build() triggers handshaking with the brokers in BootstrapServers, allocates the buffer, and starts a background I/O thread. Disposing tears all of it down.

Per-message producer (wrong)
──────────────────────────────────────────────────────────────
  Message 1  →  [Build] [TCP handshake] [Produce] [Flush] [Dispose]
  Message 2  →  [Build] [TCP handshake] [Produce] [Flush] [Dispose]
        ↑                   ↑
   Wasted allocation    Network RTT on 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. Because the queue is native memory, the GC cannot observe or collect it. 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.

<table> <thead><tr><th>🔧 What you called</th><th>📬 How an error arrives</th><th>🔍 What to inspect</th></tr></thead> <tbody> <tr><td><code>ProduceAsync</code></td><td>Throws <code>ProduceException<K,V></code></td><td><code>ex.Error</code>, <code>ex.DeliveryResult.Status</code></td></tr> <tr><td><code>Produce(..., handler)</code></td><td><code>DeliveryReport<K,V></code> passed to your handler</td><td><code>report.Error.IsError</code></td></tr> <tr><td><code>Consume</code></td><td>Throws <code>ConsumeException</code></td><td><code>ex.Error</code>, <code>ex.ConsumerRecord</code></td></tr> <tr><td>Nothing — client-level</td><td>Error handler callback, if registered</td><td><code>error.IsFatal</code>, <code>error.Code</code></td></tr> </tbody> </table>

⚠️ 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. The third value is the honest answer when an acknowledgment was lost in flight, and it's worth logging even on the success path:

try
{
    var result = await producer.ProduceAsync("orders", new Message<string, string>
    {
        Key = order.Id,
        Value = JsonSerializer.Serialize(order)
    });

    if (result.Status == PersistenceStatus.PossiblyPersisted)
    {
        logger.LogWarning(
            "Order {OrderId} may or may not have been written — treat as ambiguous",
            order.Id);
    }
}
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);
    // Route the raw bytes to a dead-letter topic rather than discarding them.
}

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 are delivered only to 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, these are invisible.
        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.LogDebug("[librdkafka] {Message}", 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 a Tight Loop Without a Cancellation Token

Consume() with no cancellation token blocks indefinitely until a message is available or a fatal error occurs. 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 forces either waiting for the next message or escalating to SIGKILL.

// ⚠️ 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     │
      │                   │                       │
      │                   │──── LeaveGroup ──────▶│
      │                   │                       │ Reassigns partitions
      │                   │◀─── Confirm ──────────│  immediately
      │                   │                       │
      │                   │ consumer.Close()      │
      │                   │ consumer.Dispose()    │
      │◀──── 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. Note that a blocking Consume directly inside ExecuteAsync will stall host startup — wrap the loop in Task.Run so ExecuteAsync returns promptly.

Mistake 4: Assuming IConsumer Is Thread-Safe

The threading models are deliberately asymmetric: IProducer.Produce and IProducer.ProduceAsync are thread-safe; IConsumer methods are not, and calling them concurrently produces undefined behavior.

The producer's safety comes from librdkafka's lock-protected internal queue. The consumer's poll-offset-commit cycle is a stateful sequence with no internal locking — fetching updates partition assignment state, and group coordination runs against the same handle. Concurrent calls produce silent data loss or incorrect offset commits.

// ⚠️ WRONG: sharing one consumer across threads
for (int i = 0; i < workerCount; i++)
{
    Task.Run(() =>
    {
        while (true)
        {
            var result = consumer.Consume(stoppingToken); // ← race condition
            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 consumer thread 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 solves thread-safety 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 duplicate and out-of-order processing are both genuinely acceptable; Consumer & Manual Commits develops the correct version.

	IProducer	IConsumer
Thread-safe methods	Produce, ProduceAsync	None
Shared across threads	Safe by design	Undefined behavior
Recommended model	Singleton, concurrent callers	One instance per thread

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, exceptions that don't trace into managed code, a background thread that persists 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 thread and internal queue are native resources, and a Flush you control with a timeout you chose is the only way to know whether pending messages made it.

Long-Lived, Stateful Objects — Not Request-Scoped

Both client types own TCP connection pools, internal queues, and background threads. Treating them as request-scoped multiplies construction cost with every operation and leaks native resources.

🧠 Mnemonic: Think of IProducer and IConsumer the way you think of HttpClient — one instance for the application lifetime, not one per call.

// ✅ 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.
    services.AddHostedService<KafkaShutdownService>();

    return services;
}

IConsumer follows the same rule with an extra consequence: 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). Guid is not on that list.

Error surfaces: Four distinct mechanisms, only two of which are exceptions on the call you made. 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

Concept	❌ Before	✅ After
Client architecture	"It's just a .NET library"	Managed wrapper over librdkafka; native memory and threads
Config defaults	Whatever the Kafka docs say	librdkafka's defaults, which differ from the Java client's
Object lifetime	Create per message / per request	Singleton for the application lifetime
Disposal	using block is enough	Flush(timeout) and check the return, then Dispose; Close before Dispose for consumers
Serialization	The client handles it	Caller-registered; no built-in for Guid, DateTime, decimal
Error handling	try/catch around the call	Four surfaces; DeliveryResult/ConsumeResult have no Error property
Cancellation	Optional	Required for graceful shutdown

What Comes Next

Producer Configuration takes the ProducerConfig populated here with only BootstrapServers and covers the knobs governing delivery guarantees and batching: acks, linger.ms, batch.size, max.in.flight.requests.per.connection, idempotence, and their interactions. It assumes you understand the internal queue and background thread, which you now do.

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.

🤔 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.