Production Patterns & Observability

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

• write a consume loop that retries transient failures without silently skipping the message, and quarantines poison records in a dead-letter topic;
• choose an idempotency key that actually catches the duplicates your pipeline produces;
• build an outbox relay that neither loses nor reorders events;
• read per-partition lag correctly and compute time to catch up from the net drain rate;
• shut a Kafka BackgroundService down cleanly and carry trace context across Kafka.

This lesson assumes "Consumer Groups & Offsets" (committed offset = next offset to read), "Raw Confluent.Kafka Client", "Consumer & Manual Commits" (the two commit pairings) and "Delivery Semantics & Schema Strategy". Each major section ends with an exercise — try it before opening the answer.

Why Production Kafka Is Different From Development Kafka

Your local Kafka setup is a polite fiction. The broker starts reliably, messages arrive in order, your consumer processes them without interruption, and the whole stack hums along on a single machine with predictable latency. It's a useful fiction — it lets you focus on business logic without fighting infrastructure — but it trains you to expect a world that production will never deliver. In production, brokers restart under you mid-consumer, partitions rebalance while you're mid-batch, network timeouts turn a five-millisecond round-trip into a thirty-second stall, and messages arrive that your deserializer has never seen and cannot parse. The gap between local and production is not a configuration gap. It is a design gap — and closing it requires understanding precisely where and why things break.

This section maps that gap. The patterns covered in the rest of this lesson — idempotent consumers, dead-letter topics (DLTs: side topics that park messages the consumer has given up on, so the partition can move on), the outbox pattern, lag alerting, clean shutdown and tracing — all exist because of the failure modes introduced here.

---

The Happy Path Is a Special Case

In development, the implicit contract looks something like this:

Produce message → Broker stores it → Consumer reads it → Processing succeeds → Done

Every component cooperates. In production, any arrow in that chain can fail, delay, or produce partial results:

  Producer          Broker             Consumer             Downstream
     │                │                   │                     │
     │──produce()────►│                   │                     │
     │                │ ← broker restart  │                     │
     │                │   or leader       │                     │
     │                │   election        │                     │
     │                │──deliver─────────►│                     │
     │                │                   │ ← rebalance fires   │
     │                │                   │   mid-batch         │
     │                │                   │──downstream ───────►│
     │                │                   │   write             │ ← timeout,
     │                │                   │                     │   conflict,
     │                │                   │                     │   or crash

Each failure point has a different character. A broker restart is transient — wait and retry, and the broker comes back. A malformed message payload is permanent — no amount of retrying will make unparseable JSON deserialize correctly. Understanding which failure you're facing is the first design decision your consumer must make.

💡 Mental Model: Think of your Kafka pipeline the way a civil engineer thinks about a bridge: each individual component can be well-built, but the system's integrity is determined by the weakest load path under stress. A producer that reliably publishes messages cannot compensate for a consumer that silently swallows errors. Production readiness belongs to the full pipeline, not to any single component.

---

Brokers Restart, Partitions Rebalance, Timeouts Happen

These three events are not edge cases in a well-run Kafka cluster. They are routine operational events. Rolling upgrades restart brokers one at a time. Consumer group membership changes trigger rebalances. Cloud networking introduces variable latency that occasionally crosses your client's timeout thresholds.

The critical implication: your consumer will be interrupted mid-processing. The message your consumer is currently working on may have its partition reassigned to a different consumer instance before you commit the offset. If the process dies or loses the partition before the commit, whoever owns the partition next reads that message again — possibly many times.

Here's a minimal consumer loop in C# that exposes where the interruption risk lives:

// Simplified to illustrate timing — production patterns shown in later sections
using Confluent.Kafka;

var config = new ConsumerConfig
{
    BootstrapServers = "kafka:9092",
    GroupId = "order-processor",
    AutoOffsetReset = AutoOffsetReset.Earliest,
    EnableAutoCommit = false  // Manual commit gives us control over timing
};

using var consumer = new ConsumerBuilder<string, string>(config).Build();
consumer.Subscribe("orders");

while (!cancellationToken.IsCancellationRequested)
{
    // ① Message received
    var result = consumer.Consume(cancellationToken);

    // ② Processing happens here — rebalance or crash can interrupt at any point
    await ProcessOrderAsync(result.Message.Value);

    // ③ Offset committed AFTER successful processing
    //    If the process crashes between ② and ③, the message is redelivered
    //    Both behaviors are intentional — they preserve at-least-once delivery
    consumer.Commit(result);
}

A crash between receiving the message and committing its offset means Kafka will redeliver the message when the consumer restarts or when the partition is reassigned. That's not a bug — it's the at-least-once delivery guarantee in action. But it means your processing logic must be safe to run more than once on the same message.

🤔 Did you know? A partition rebalance invokes the PartitionsRevoked callback on your consumer — the last opportunity to commit the offsets of work you have finished before the partitions are handed to another instance. With manual commits, anything finished but not yet committed is reprocessed by the new owner. (With auto-commit on, librdkafka commits the stored offsets on revocation by itself.) Confluent.Kafka exposes it via ConsumerBuilder.SetPartitionsRevokedHandler; partitions that are lost rather than revoked go to SetPartitionsLostHandler, and those can no longer be committed. Mistake 3 below shows a handler that commits only finished work.

---

At-Least-Once Is the Usual Target — and the Starting Design Constraint

At-least-once delivery means every produced message is processed at least once, but potentially more than once. Kafka does not give you that by itself: it takes a producer that doesn't lose writes (acks=all) and a consumer that commits only after processing. Confluent.Kafka's consumer defaults (auto-commit plus automatic offset storing) do not guarantee it — they can commit past a message your handler never finished. At-least-once is the right target for most use cases; the alternative (at-most-once) means accepting data loss on failure.

But at-least-once puts the burden of duplicate safety on your consumer.

📋 Delivery Semantics Comparison

┌─────────────────────┬──────────────────────┬───────────────────────┐
│ Semantic            │ Data Loss Risk       │ Duplication Risk      │
├─────────────────────┼──────────────────────┼───────────────────────┤
│ At-most-once        │ 🔴 Yes (on failure)   │ ✅ None               │
│ At-least-once       │ ✅ None               │ 🟡 Yes (on retry)     │
│ Exactly-once        │ ✅ None               │ ✅ None (within Kafka)│
└─────────────────────┴──────────────────────┴───────────────────────┘

Note: Exactly-once semantics apply within the Kafka broker and transactional
producer/consumer pair. They do NOT automatically extend to your external
downstream systems (databases, APIs, caches).

❌ Wrong thinking: "I'll configure enable.idempotence=true on the producer and exactly-once will take care of itself."

✅ Correct thinking: Producer-side idempotence prevents duplicate broker writes from the client's own internal retries, but your consumer processing the same message twice — due to a rebalance, a crash, or a retry — is a separate concern that requires deliberate consumer-side design.

---

Transient Errors vs. Poison-Pill Messages

Not all failures in a consumer loop are the same kind of problem. The distinction that matters most is between a transient error and a poison-pill message.

A transient error is a failure likely to resolve on its own — a temporarily overloaded database, a network timeout, a momentary loss of connectivity. The right response is to retry with backoff.

A poison-pill message will never succeed regardless of how many times you retry it — malformed JSON, an incompatible schema version, a reference to an entity ID that doesn't exist. Retrying a poison pill doesn't fix anything; it blocks the partition and stops all forward progress on every subsequent message behind it.

// Illustrating the two different failure branches — full DLT routing
// is covered in 'Idempotent Consumers and the Dead-Letter Topic Contract'.
try
{
    var order = JsonSerializer.Deserialize<OrderEvent>(result.Message.Value);
    await SaveToDatabase(order);
    consumer.Commit(result);
}
catch (JsonException ex)
{
    // Poison pill: no retry will fix a malformed payload.
    _logger.LogError(ex, "Undeserializable message at {Offset}", result.TopicPartitionOffset);
    await RouteToDeadLetterAsync(result, ex, attempts: 1);   // awaits the DLT acknowledgement
    consumer.Commit(result);
}
catch (DbException ex) when (ex.IsTransient)
{
    // Transient error. Consume() has ALREADY moved this partition's position past
    // result.Offset, so merely not committing does not bring the message back: the
    // next Consume() would return the NEXT message, and committing that one
    // (offset + 1) would silently skip this one. Rewind explicitly instead.
    _logger.LogWarning(ex, "Transient DB error, will retry {Offset}", result.TopicPartitionOffset);
    await Task.Delay(retryDelay, cancellationToken);
    consumer.Seek(result.TopicPartitionOffset);   // the next Consume() re-reads this message
    // (Unbounded here for brevity: "Error Routing" below caps the attempts.)
}

🎯 Key Principle: Route transient errors toward retry — and make the retry real: retry in-process, or Seek back to the failed offset. Merely not committing does not bring the message back while the consumer keeps running, because Consume has already moved past it. Route permanent errors toward dead-letter topics (preserve the message, then commit past it). Confusing the two produces either an infinite retry loop or silent data loss — both are production outages.

---

Production Readiness Is a Pipeline Property

The failure modes that cause real incidents almost always span component boundaries:

┌──────────────────────────────────────────────────────────────────────────┐
│                   Cross-Boundary Failure Scenarios                       │
├─────────────────────┬────────────────────────────────────────────────────┤
│ 🔒 Producer + DB    │ Produce succeeds, DB transaction rolls back →      │
│                     │ event published for business state that never      │
│                     │ committed (the dual-write problem)                 │
├─────────────────────┼────────────────────────────────────────────────────┤
│ 🔁 Broker + Consumer│ Broker delivers message, consumer processes it,    │
│                     │ broker triggers rebalance before commit →          │
│                     │ message redelivered and processed twice            │
├─────────────────────┼────────────────────────────────────────────────────┤
│ 📉 Consumer + Down- │ Consumer commits offset, downstream write fails    │
│    stream           │ silently → message consumed but effect lost        │
├─────────────────────┼────────────────────────────────────────────────────┤
│ 🕳️ All components   │ Lag grows silently, no alert fires, backlog of     │
│                     │ unprocessed messages accumulates undetected        │
└─────────────────────┴────────────────────────────────────────────────────┘

Each row is a scenario where individual components function correctly but the system as a whole is failing. The outbox pattern (writing the event to an outbox table in the same database transaction as the business change, then publishing it to Kafka from there) addresses the producer-to-database boundary. Consumer lag alerting addresses the detection gap where a pipeline can fail silently without any single component throwing an error.

The rest of this lesson builds directly on these failure modes:

• 🔁 Redelivery and at-least-once semantics → idempotent consumer design and offset management
• 💀 Poison-pill messages → dead-letter topics that don't block partition progress
• 📉 Silent pipeline failures → consumer lag measurement and alerting
• 🔒 The dual-write problem → the outbox pattern

✍️ Exercise: transient, permanent, or not your loop's problem?

For each failure, say whether the consume loop should retry, dead-letter, or do nothing special, and why.

1. A partition leader election happens while your producer is sending.
2. An OrderPlaced payload is missing its required orderId field.
3. The inventory API returns HTTP 503 for ten seconds.
4. An order references a customer ID that was never created and never will be.
5. Consume throws a ConsumeException with ErrorCode.Local_ValueDeserialization.

Check your answer

1. Nothing special. The client library handles it: it refreshes metadata and retries internally. Your code sees at most a slower ProduceAsync.
2. Dead-letter. The payload will be just as invalid on the hundredth attempt.
3. Retry with backoff, bounded: this is the textbook transient failure. Keep the total retry time well under max.poll.interval.ms, or the consumer is evicted from the group.
4. Dead-letter. It is a business-level poison pill: retrying cannot create the customer. (If the customer might arrive a little later, because of an ordering issue upstream, a short bounded retry first is reasonable.)
5. Dead-letter the raw bytes. The exception's ConsumerRecord still carries the key and value bytes, and the consumer's position has already moved past the record, so if you only log it, it is silently skipped.

---

Idempotent Consumers and the Dead-Letter Topic Contract

With those failure modes in mind, two mechanisms close the gap between Kafka's delivery guarantee and the safety your downstream systems require: idempotent consumers and the dead-letter topic (DLT). Neither is complicated in isolation, but the interaction between them — particularly around offset commits — is where most production incidents originate.

The Idempotency Key: Your Defense Against Duplicate Processing

An idempotent consumer produces the same observable outcome whether it processes a given message once or a hundred times. The mechanism: before acting on a message, check whether you have already processed it; if you have, skip it and commit the offset as if you had processed it normally. The check depends on an idempotency key — a value that uniquely identifies the message across deliveries.

You have two options for sourcing this key:

• 🔑 Natural key: a domain-level identifier embedded in the message payload — an OrderId, a PaymentTransactionId, a DeduplicationToken. Preferred when the upstream system can guarantee a stable unique identifier per business event.
• 🔧 Synthetic key: the Kafka coordinates themselves — topic + partition + offset. A record keeps the same topic-partition-offset triple for its whole life, so this tuple reliably catches redelivery of the same record — the at-least-once case — even when the payload carries no domain identifier. It cannot catch a duplicate the producer wrote twice (for example an outbox relay re-publishing after a crash): the second copy is a different record at a different offset. For those you need an event ID assigned once, before the first send — the outbox row's ID, carried in a header or the payload.

The key must be stored in a durable store — a database, Redis, or any system that survives a consumer restart. An in-memory set works locally but is useless in production because a pod restart empties it, and the very restart that triggers redelivery also clears your deduplication record.

┌─────────────────────────────────────────────────────┐
│                  Consumer Process                   │
│                                                     │
│  Receive message (topic, partition, offset, payload)│
│          │                                          │
│          ▼                                          │
│  ┌───────────────┐   key found   ┌───────────────┐  │
│  │ Check durable │──────────────►│ Skip + Commit │  │
│  │ idempotency   │               │    offset     │  │
│  │    store      │               └───────────────┘  │
│  └───────┬───────┘                                  │
│          │ key NOT found                            │
│          ▼                                          │
│  ┌───────────────┐               ┌───────────────┐  │
│  │ Execute       │──────────────►│ Write key to  │  │
│  │ downstream    │    success    │ store, Commit │  │
│  │ business      │               │    offset     │  │
│  │ logic         │               └───────────────┘  │
│  └───────┬───────┘                                  │
│          │ failure                                  │
│          ▼                                          │
│  ┌───────────────┐                                  │
│  │ Retry or      │                                  │
│  │ route to DLT  │                                  │
│  └───────────────┘                                  │
└─────────────────────────────────────────────────────┘

⚠️ Critical ordering: Write the idempotency key in the same atomic operation as the downstream write, or after it — never before. If you write the key first and then crash before the downstream write completes, a retry will find the key, believe the message was processed, and silently discard the event. "After" is only safe when the downstream write is itself repeatable (an upsert): a crash between the write and the key write means the write runs again.

The Commit Tradeoff: At-Least-Once vs. At-Most-Once

The position of the commit relative to the downstream write determines your delivery semantics:

At-least-once with idempotent consumers is the pattern almost all production Kafka systems target. The Confluent .NET client exposes two methods for precise control:

• Commit(consumeResult) — synchronously commits offset + 1 for that partition (the committed offset is always the next offset to read).
• StoreOffset(consumeResult) — records offset + 1 in the client's in-memory offset store; nothing is sent to the broker. The background auto-committer (or a parameterless Commit()) sends stored offsets later. It requires EnableAutoOffsetStore = false — with the default (true) the call throws.

Use one of two pairings: (a) EnableAutoCommit = false plus Commit(result) after processing, or (b) EnableAutoCommit = true plus EnableAutoOffsetStore = false plus StoreOffset(result) after processing. With the defaults (both automatic), the client stores each offset the moment Consume hands you the message and commits it on a timer, whether or not processing succeeded. The loops in this lesson use pairing (a).

var config = new ConsumerConfig
{
    BootstrapServers = "localhost:9092",
    GroupId = "order-processor",
    AutoOffsetReset = AutoOffsetReset.Earliest,
    EnableAutoCommit = false          // pairing (a): explicit Commit after processing
};

using var consumer = new ConsumerBuilder<string, string>(config).Build();
consumer.Subscribe("orders");

while (!cancellationToken.IsCancellationRequested)
{
    ConsumeResult<string, string> result;
    try
    {
        result = consumer.Consume(cancellationToken);
    }
    catch (ConsumeException ex)
    {
        // A per-record failure — usually a key/value deserialization error. The
        // position has already moved past the record, so `continue` skips it for
        // good; real code dead-letters ex.ConsumerRecord's raw bytes first
        // (see "Error Routing" below).
        logger.LogError(ex, "Consume error {Code} at {Position}",
            ex.Error.Code, ex.ConsumerRecord?.TopicPartitionOffset);
        continue;
    }

    var idempotencyKey = $"{result.Topic}:{result.Partition.Value}:{result.Offset.Value}";

    if (await idempotencyStore.ExistsAsync(idempotencyKey))
    {
        consumer.Commit(result);      // already handled: just move past it
        continue;
    }

    await ProcessOrderAsync(result.Message.Value);
    await idempotencyStore.SetAsync(idempotencyKey, TimeSpan.FromDays(7));

    consumer.Commit(result);
}

The key uses Partition.Value and Offset.Value: Partition.ToString() renders as [0], which would put brackets into every key. Keep the key's time-to-live longer than the longest replay you might run (at least the topic's retention).

The Dead-Letter Topic: Quarantine Without Blocking

A poison-pill message — malformed payload, invalid schema, unrecoverable business logic exception — will fail on every retry. Without a safety valve, your consumer stalls at that offset indefinitely, blocking every message behind it on that partition. This is head-of-line blocking, one of the most disruptive failure modes in production Kafka.

The dead-letter topic (DLT) is the safety valve. The contract: once a message is judged unprocessable (at once for a permanent failure, or when a transient one has used up its bounded retries), route it to a separate topic (conventionally <original-topic>.DLT) and commit the original offset, allowing the main consumer to advance. The message is preserved for operator inspection and eventual replay.

Main Topic: orders
─────────────────────────────────────────────────────
  [msg A] [msg B] [msg C ← poison pill] [msg D] [msg E]
                       │
                       │  fails permanently, or
                       │  exhausts its retries
                       ▼
             ┌─────────────────┐
             │  Unprocessable: │
             │  quarantine it  │
             └────────┬────────┘
                      │
          ┌───────────┴───────────┐
          ▼                       ▼
  DLT Topic:               Original offset
  orders.DLT               committed → msg D
  [msg C + headers:         and E can now
   error reason,            be processed
   retry count,
   original offset]

The ordering constraint is strict: write to the DLT and wait for the broker's acknowledgement before committing the original offset. If you commit first and the DLT write fails, the message is never processed and never inspected: the group has moved past it and the DLT never received it. If the DLT write succeeds but the commit crashes, you may write to the DLT twice on replay — that is recoverable.

"Wait for the acknowledgement" needs care. A fire-and-forget Produce followed by Flush does not prove the write succeeded: Flush only returns how many messages are still queued, and a delivery failure goes to a delivery handler that nobody is checking. Awaiting ProduceAsync does prove it — it completes when the broker acknowledges and throws a ProduceException when it cannot:

private async Task RouteToDeadLetterAsync(
    ConsumeResult<string, string> failed, Exception reason, int attempts)
{
    var dltMessage = new Message<string, string>
    {
        Key = failed.Message.Key,
        Value = failed.Message.Value,
        Headers = DeadLetterHeaders(failed.Message.Headers, failed.TopicPartitionOffset, reason, attempts)
    };

    // ProduceAsync completes only once the broker has acknowledged the write,
    // and throws ProduceException if it could not — so no separate Flush is
    // needed, and a failed DLT write can never be followed by a commit.
    await _dltProducer.ProduceAsync($"{failed.Topic}.DLT", dltMessage);

    // The caller commits the original offset only after this returns.
}

// Text (UTF-8) values keep the headers readable by any tool or language.
private static Headers DeadLetterHeaders(
    Headers? original, TopicPartitionOffset origin, Exception reason, int attempts)
{
    static byte[] Text(string s) => Encoding.UTF8.GetBytes(s);
    var inv = CultureInfo.InvariantCulture;

    // Keep the original headers (event-id, traceparent), so a message replayed from
    // the DLT can still be deduplicated and traced. The dlt.* entries go last, so
    // GetLastBytes reads this failure's provenance.
    var headers = new Headers();
    foreach (var h in original ?? new Headers()) headers.Add(h.Key, h.GetValueBytes());
    headers.Add("dlt.original.topic",     Text(origin.Topic));
    headers.Add("dlt.original.partition", Text(origin.Partition.Value.ToString(inv)));
    headers.Add("dlt.original.offset",    Text(origin.Offset.Value.ToString(inv)));
    headers.Add("dlt.error.type",         Text(reason.GetType().FullName ?? reason.GetType().Name));
    headers.Add("dlt.error.reason",       Text(reason.Message));
    headers.Add("dlt.attempts",           Text(attempts.ToString(inv)));
    headers.Add("dlt.timestamp.utc",      Text(DateTime.UtcNow.ToString("O", inv)));
    return headers;
}

The headers are critical — without them the DLT is a graveyard rather than a quarantine: you have the message but no cause of death.

The Outbox Pattern: Solving the Dual-Write Problem

Neither idempotency nor the DLT addresses the gap between writing to your database and producing to Kafka. The naive pattern:

1. BEGIN database transaction
2. Write business record to DB
3. COMMIT database transaction
4. Produce event to Kafka

If the process crashes after step 3 and before step 4, the database reflects a state change that Kafka never announced. This is the dual-write problem: two writes to two systems that cannot commit together. It is insidious because it produces no error: the first write succeeded, and nothing records that the second never happened.

The outbox pattern eliminates the problem by writing the event to a table in the same database transaction as the business record. Both writes share an ACID transaction — they either both commit or both roll back.

┌─────────────────────────────────────────────────────┐
│               Application Database                  │
│                                                     │
│  ┌──────────────────┐    same transaction           │
│  │  orders table    │◄──────────────────────┐       │
│  │  (business data) │                       │       │
│  └──────────────────┘            ┌──────────┴─────┐ │
│                                  │  Application   │ │
│  ┌──────────────────┐            │  writes both   │ │
│  │  outbox table    │◄───────────│  atomically    │ │
│  │  (pending events)│            └────────────────┘ │
│  └────────┬─────────┘                               │
└───────────┼─────────────────────────────────────────┘
            │ polled by relay worker
            ▼
┌─────────────────────┐        ┌──────────────────┐
│  Outbox Relay       │──────► │  Kafka Broker    │
│  BackgroundService  │  ack   │  (orders topic)  │
│  (polls + produces) │◄───────│                  │
└─────────────────────┘        └──────────────────┘
     deletes row only
     after ack received

The outbox relay worker polls the outbox table, produces each unpublished row to Kafka, and deletes the row only after receiving a broker acknowledgment. If it crashes after producing but before deleting, it will produce the event again on restart. That second copy is a new record at a new offset, so the receiving consumer must deduplicate on the event's own ID — the outbox row ID, sent as a header — not on topic-partition-offset. With that, the reliability loop is complete.

The relay producer should also be configured with EnableIdempotence = true. That stops the client's own internal retries from writing a batch twice or out of order within one producer session. It cannot recognise a row the relay produces again after a timeout or a restart: to the broker that is a new record with a new sequence number (and, after a restart, a new producer ID). It complements consumer-side deduplication; it does not replace it.

💡 Real-World Example: An e-commerce checkout service writes the Order record to a relational database and must emit an OrderPlaced event to Kafka. Without the outbox pattern, a container eviction between the DB commit and the Kafka produce means the order exists in the database but no downstream service knows about it — the warehouse never packs the box, the customer never gets a confirmation email. With the outbox pattern, the OrderPlaced row survives any crash and the relay eventually delivers it.

Connecting the Mechanisms

Idempotency, offset management, the DLT contract, and the outbox pattern form a pipeline of guarantees:

The practical C# code that assembles these mechanisms into a working consumer loop is in "Practical Patterns in .NET: Putting It Together", after the lag section.

✍️ Exercise: which duplicates does your key catch?

An outbox row with ID 7f3c… carries OrderPlaced for order 1001. The relay's ProduceAsync times out, although the broker did write the record, at offset 880. The relay crashes before deleting the row, restarts, and publishes the row again, this time at offset 912. The warehouse consumer deduplicates with the key orders:{partition}:{offset}.

1. How many times does the warehouse pack order 1001?
2. Would EnableIdempotence = true on the relay have prevented the second copy?
3. What change makes the warehouse pack it once?

Check your answer

1. Twice. Offsets 880 and 912 give two different keys, so both copies look new.
2. No. The restarted relay has a new producer ID and the re-send has a new sequence number, so the broker sees an unrelated record. Idempotence only covers the client's own retries within one session.
3. Deduplicate on the event's identity instead: send the outbox row ID as an event-id header (or put it in the payload), and record it in the warehouse's dedupe store in the same transaction as the packing record. Both copies then carry 7f3c…, and the second is skipped.

---

Consumer Lag: What It Measures and How to Act on It

With the reliability mechanisms in place, you need a way to know when the pipeline is under stress before it causes a customer-facing incident. Consumer lag is that signal. It is the earliest warning that your pipeline is struggling, and it is the metric most likely to give you time to act before a backlog becomes an outage. But lag is also routinely misread — teams set alerts on raw offset counts and either panic unnecessarily or miss the slow growth that signals genuine capacity exhaustion.

What Lag Actually Measures

Consumer lag is the difference between the log-end offset (the offset of the next message that will be written to a partition) and the committed offset (the offset of the next message the group will read — its last processed offset + 1). That subtraction happens at the partition level, not the topic level — a topic with 12 partitions has 12 independent lag values, each potentially telling a different story.

Partition 0:  log-end offset = 10,450   committed offset = 10,448   lag = 2
Partition 1:  log-end offset = 10,501   committed offset = 9,870    lag = 631
Partition 2:  log-end offset = 10,389   committed offset = 10,387   lag = 2

An aggregate "topic lag" of 635 sounds alarming, but partition 1 carries almost all of it — which immediately tells you to look at the consumer instance assigned to that partition, not the consumer group as a whole. Per-partition lag is the unit of diagnosis; a rolled-up sum is useful only as a high-level health indicator.

Reading the Shape of Lag, Not Just the Number

The pattern of lag over time is more diagnostic than any single reading. Two shapes appear repeatedly in production:

Steady, monotonically growing lag means the consumer is processing messages more slowly than the producer is writing them. This is a capacity problem: the consumer group lacks throughput to keep up, and the solution involves horizontal scaling, throughput optimization, or reducing per-message processing cost.

Time →
Lag:  100 ... 800 ... 1,500 ... 2,200 ... 2,900
                ↑ steady (linear) growth = consumers fall behind at a constant rate

A constant capacity gap grows lag linearly: here by 700 per interval. A curve that accelerates means the gap is widening: the consume rate is falling (a downstream dependency getting slower, for example) or the produce rate is rising.

Spike-and-recover lag looks different — lag jumps sharply, then returns to near-zero. This pattern almost never indicates capacity problems. It is the signature of a consumer group rebalance, a short burst of producer traffic, or a processing slowdown on a single expensive message. With the classic eager protocol (librdkafka's default range,roundrobin assignment), every member stops consuming during a rebalance. Cooperative-sticky assignment and the newer group.protocol=consumer (KIP-848) pause only the partitions that actually move.

|         ___
|        /   \
|       /     \
|______/       \_____________
time→   ^ rebalance  ^ recovery
        starts

A single partition whose committed offset stops moving while its lag climbs at the producers' rate — while the group's other partitions drain normally — is the clearest indicator of a message that needs routing to a dead-letter topic. The correct response is to get the unprocessable message out of the way (the DLT) — not to increase max.poll.interval.ms, which only delays the moment a stuck loop is evicted from the group.

Where Lag Numbers Come From

Lag tools build on the same two broker-side numbers: committed offsets, which live in the internal __consumer_offsets topic and are read through the group coordinator's OffsetFetch API, and log-end offsets, read through the ListOffsets API. (librdkafka also computes a client-side consumer_lag per partition in its statistics — see "A dashboard that tells the truth" below.)

In .NET, Confluent.Kafka exposes this through IAdminClient:

using Confluent.Kafka;
using Confluent.Kafka.Admin;

var config = new AdminClientConfig { BootstrapServers = "localhost:9092" };
using var adminClient = new AdminClientBuilder(config).Build();

var partitions = new List<TopicPartition>
{
    new TopicPartition("orders", new Partition(0)),
    new TopicPartition("orders", new Partition(1)),
};

// Committed offsets for the group (the OffsetFetch API)
var groupResults = await adminClient.ListConsumerGroupOffsetsAsync(
    new[] { new ConsumerGroupTopicPartitions("my-consumer-group", partitions) });
var committed = groupResults.Single().Partitions;

// Log-end offsets for the same partitions (the ListOffsets API)
var logEnd = await adminClient.ListOffsetsAsync(
    partitions.Select(tp => new TopicPartitionOffsetSpec
    {
        TopicPartition = tp,
        OffsetSpec = OffsetSpec.Latest()
    }));

foreach (var c in committed)
{
    var end = logEnd.ResultInfos
        .Select(r => r.TopicPartitionOffsetError)
        .First(r => r.TopicPartition == c.TopicPartition);

    // A partition the group has never committed reports Offset.Unset (-1001);
    // subtracting that would print a meaningless lag.
    if (c.Offset == Offset.Unset)
    {
        Console.WriteLine($"{c.TopicPartition}: no committed offset yet");
        continue;
    }

    Console.WriteLine($"{c.TopicPartition}: lag = {end.Offset.Value - c.Offset.Value}");
}

(Checked against Confluent.Kafka 2.15: TopicPartitionOffsetSpec is set through its properties, OffsetSpec.Latest() is a method, and ListOffsetsAsync returns a ListOffsetsResult whose ResultInfos hold the offsets.)

In a real monitoring service you would emit these values on a polling interval as metrics (gauges, which report a current value) through OpenTelemetry (the vendor-neutral standard for traces, metrics and logs) or straight to a metrics system such as Prometheus, rather than printing them.

🤔 Did you know? When a consumer group has no live members — scaled to zero, crashed, or stopped — its committed offsets remain frozen in __consumer_offsets while the log-end offset keeps advancing. Lag grows silently with no consumer running. A lag alert will fire correctly in this case — as long as messages keep arriving, and as long as lag is computed by something other than the dead consumer. Mistake 4 covers the cases where it doesn't. Watch group liveness as a separate signal; both belong on your dashboard.

Alerting in Time-to-Catch-Up, Not Raw Offsets

A threshold of "alert if lag exceeds 10,000 messages" immediately runs into a problem: 10,000 offsets represents two seconds of work on a high-throughput topic and three hours of backlog on a low-throughput one. The same number is both too sensitive and not sensitive enough.

The fix is to express alerting thresholds in time-to-catch-up: how long the existing lag takes to drain at current rates. Producers keep writing while consumers drain, so what shrinks the backlog is the net rate:

time_to_catch_up_seconds = current_lag / (consume_rate − produce_rate)
                           (only defined when consume_rate > produce_rate)

A lag of 50,000 with consumers at 5,000 msg/s and producers at 4,000 msg/s drains at a net 1,000 msg/s: 50 seconds. The same lag with producers also at 5,000 msg/s never drains, even though lag / consume_rate would claim 10 seconds. Both rates come from the same two numbers you already collect: the consume rate is how fast the committed offset rises, and the produce rate is how fast the log-end offset rises.

// Illustrative lag metric enrichment
public record LagMetric(
    TopicPartition TopicPartition,
    long CurrentLag,
    double ConsumeRatePerSecond,   // how fast the committed offset is rising
    double ProduceRatePerSecond)   // how fast the log-end offset is rising
{
    // Null when the backlog is not shrinking (net drain rate <= 0): at current
    // rates the lag never clears. Harmless if it is small and steady; alarming if it grows.
    public double? TimeToCatchUpSeconds =>
        ConsumeRatePerSecond > ProduceRatePerSecond
            ? CurrentLag / (ConsumeRatePerSecond - ProduceRatePerSecond)
            : null;
}

// Alert when TimeToCatchUpSeconds exceeds your SLO tolerance, or is null while lag keeps growing
// e.g., > 300 seconds = warning; > 900 seconds, or null with lag rising for 10 minutes = page on-call

From Alert to Action: The Runbook Contract

A lag alert with no defined response is noise. The value of consumer lag as a signal comes from pairing it with concrete, pre-agreed responses tied to the pattern of lag:

For horizontal scaling, the partition count is the ceiling on parallelism — you cannot usefully add more consumer instances than there are partitions in a consumer group. (Kafka 4.2's share groups lift that limit for workloads that don't need per-partition ordering; check your client's support before relying on them.) If you are already at partition count and lag is still growing, the next lever is reducing per-message processing cost or adding partitions (which requires careful planning, as it changes key routing).

Increasing max.poll.interval.ms is legitimate only when lag spikes correlate with genuinely long-running processing on non-poisoned messages. It also means a consumer whose loop is stuck (alive, but not polling) takes longer to be evicted — set it only as high as your worst-case legitimate processing time. A crashed process is detected separately, through missed heartbeats (session.timeout.ms, 45 seconds by default in librdkafka).

💡 Real-World Example: A team running an order-enrichment consumer noticed lag creeping up on all partitions each Tuesday afternoon, then recovering by end of day. Investigation revealed a downstream inventory API ran a heavy report job every Tuesday, adding ~400ms of latency per call. The fix was a per-call timeout behind a circuit breaker (which stops calling a struggling dependency for a cool-off period), and routing messages to a retry topic (a side topic that a separate consumer retries later, so the main partition keeps moving) when the API was slow — keeping per-message processing time bounded. The time-to-catch-up metric made the pattern visible; the runbook directed the team to check downstream latency rather than immediately scaling.

Instrumenting Lag in Your .NET Service

The following BackgroundService polls the admin API every 15 seconds and exposes per-partition lag as an OpenTelemetry observable gauge (a gauge whose value is read through a callback each time metrics are collected). Run it in a separate exporter service, not inside the consumer it measures: if the consumer process dies, a lag reporter living in the same process dies with it, and the lag series disappears instead of rising — Mistake 4's blind spot. (The metric name is this example's own; name it to fit your conventions.)

using System.Collections.Concurrent;
using System.Diagnostics.Metrics;
using Confluent.Kafka;
using Confluent.Kafka.Admin;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;

// PartitionCount is fixed at startup: if partitions are added, re-read it
// (IAdminClient.GetMetadata), or the new ones go unmeasured.
public sealed record LagReporterOptions(string GroupId, string Topic, int PartitionCount);

public sealed class LagReporterService : BackgroundService
{
    private static readonly Meter Meter = new("Orders.Kafka");
    private readonly IAdminClient _admin;
    private readonly LagReporterOptions _options;
    private readonly ILogger<LagReporterService> _logger;
    private readonly ConcurrentDictionary<int, long> _lag = new();

    public LagReporterService(
        IAdminClient admin, LagReporterOptions options, ILogger<LagReporterService> logger)
    {
        _admin = admin;
        _options = options;
        _logger = logger;
        Meter.CreateObservableGauge(
            "orders.consumer.lag", ObserveLag, unit: "{message}",
            description: "Log-end offset minus committed offset, per partition");
    }

    private IEnumerable<Measurement<long>> ObserveLag()
    {
        foreach (var (partition, lag) in _lag)
            yield return new Measurement<long>(lag,
                new KeyValuePair<string, object?>("topic", _options.Topic),
                new KeyValuePair<string, object?>("partition", partition),
                new KeyValuePair<string, object?>("group", _options.GroupId));
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        var partitions = Enumerable.Range(0, _options.PartitionCount)
            .Select(p => new TopicPartition(_options.Topic, new Partition(p)))
            .ToList();

        while (!stoppingToken.IsCancellationRequested)
        {
            try
            {
                var committed = (await _admin.ListConsumerGroupOffsetsAsync(
                        new[] { new ConsumerGroupTopicPartitions(_options.GroupId, partitions) }))
                    .Single().Partitions;

                var logEnd = (await _admin.ListOffsetsAsync(partitions.Select(tp =>
                        new TopicPartitionOffsetSpec { TopicPartition = tp, OffsetSpec = OffsetSpec.Latest() })))
                    .ResultInfos
                    .Select(r => r.TopicPartitionOffsetError)
                    .ToDictionary(r => r.Partition.Value, r => r.Offset.Value);

                foreach (var c in committed.Where(c => c.Offset != Offset.Unset))
                    _lag[c.Partition.Value] = logEnd[c.Partition.Value] - c.Offset.Value;
            }
            catch (Exception ex) when (ex is not OperationCanceledException)
            {
                // Keep the last values and log the failure. Frozen values look like a
                // flat line, so also export the time of the last successful poll and
                // alert when it goes stale: a silent exporter is its own blind spot.
                _logger.LogWarning(ex, "Lag poll failed");
            }

            await Task.Delay(TimeSpan.FromSeconds(15), stoppingToken);
        }
    }
}

// Registration (Program.cs):
//   builder.Services.AddSingleton(new LagReporterOptions("order-processor", "orders", 3));
//   builder.Services.AddSingleton(_ => new AdminClientBuilder(adminConfig).Build());
//   builder.Services.AddHostedService<LagReporterService>();

⚠️ Common Mistake: Emitting only a topic-level aggregate lag metric. When partition 1 is stalled and partitions 0 and 2 are healthy, an aggregate hides which partition is stuck — and a sum dominated by one partition looks like a group-wide capacity problem. Always emit per-partition lag, and sum at the dashboard level if you need a health overview.

✍️ Exercise: diagnose three partitions

A lag exporter records these values 60 seconds apart for the orders group:

For each partition, work out the lag at t=60 s, the produce and consume rates, and the time to catch up. Then say which partition needs action, and what lag / consume_rate would have told you about partition 1.

Check your answer

• P0: lag 10 → 5. Produce 6,000/60 = 100 msg/s, consume 6,005/60 ≈ 100.1 msg/s, a net drain of 5/60 msg/s, so time to catch up = 5 ÷ (5/60) = 60 s. It is keeping up; nothing to do.
• P1: lag 6,000 → 3,000. Produce 100 msg/s, consume 9,000/60 = 150 msg/s, net drain 50 msg/s, so time to catch up = 3,000 / 50 = 60 s. A backlog clearing on its own, typical after a rebalance or a restart. No action.
• P2: lag 1,400 → 7,400. Produce 100 msg/s, consume 0, and the committed offset is frozen at 18,500. The time to catch up is undefined (it will never clear), and lag grows by 100 every second. This is the poison-pill signature: inspect the record at offset 18,500 and dead-letter it.

The naive lag / consume_rate gives P1 3,000 / 150 = 20 s, three times too optimistic, because it ignores the 100 msg/s still arriving.

Harder variation: suppose instead that all three partitions show consume 90 msg/s against produce 100 msg/s. Every partition's lag grows by 10 msg/s (600 a minute), linearly: a capacity problem, not a poison pill. You can scale out only up to three consumers, one per partition. If you are already there, the levers are cheaper per-message processing, or more partitions (which re-maps keys, so plan it).

---

Practical Patterns in .NET: Putting It Together

The concepts covered so far — idempotency, dead-letter routing, offset management, and the outbox pattern — only pay off when implemented correctly. Theory diverges from practice most sharply in the consume loop: the single while block where every production failure mode eventually surfaces. This section walks through the concrete C# patterns that translate those concepts into working Confluent.Kafka code, with particular attention to the ordering of operations.

Manual Offset Control: The Foundation Everything Else Builds On

Auto-commit is Confluent.Kafka's default: the library stores each message's offset as soon as Consume hands it to you, and commits the stored offsets every 5 seconds on a timer, independently of whether processing succeeded. A message can fail and still have its offset advanced before you notice. Manual offset commit (EnableAutoCommit = false combined with explicit consumer.Commit(consumeResult)) makes the commit a deliberate act you control — the prerequisite for every other pattern in this section.

var config = new ConsumerConfig
{
    BootstrapServers = "localhost:9092",
    GroupId = "order-processor",
    AutoOffsetReset = AutoOffsetReset.Earliest,
    EnableAutoCommit = false
};

using var consumer = new ConsumerBuilder<string, string>(config).Build();
consumer.Subscribe("orders");

while (!stoppingToken.IsCancellationRequested)
{
    var result = consumer.Consume(TimeSpan.FromSeconds(1));
    if (result is null) continue;

    try
    {
        await ProcessMessageAsync(result);
        consumer.Commit(result); // commit only after successful processing
    }
    catch (Exception ex)
    {
        // Do NOT commit. Rethrowing ends the loop (and, by default, the host);
        // "Error Routing" below replaces this with explicit retry and DLT handling.
        logger.LogError(ex, "Processing failed for {TopicPartitionOffset}",
            result.TopicPartitionOffset);
        throw;
    }
}

💡 Pro Tip: consumer.Commit(consumeResult) commits that partition's offset + 1 — exactly the message you finished. Passing the ConsumeResult directly is safer than the parameterless overload, which commits the client's stored offsets for all assigned partitions: with the default automatic offset store, those can include messages Consume has handed out but you have not yet processed.

The Idempotent Consumer Loop

Manual commit prevents lost offsets but does not prevent duplicate side effects. Every consumer restart replays from the last committed offset — if processing is not idempotent, those replays corrupt downstream state.

// Returns normally when the message is handled (or was already handled).
// Committing is left to the caller, so a commit failure can never be
// mistaken for an application failure and dead-lettered.
async Task ProcessWithIdempotencyAsync(ConsumeResult<string, string> result, IDatabase redis)
{
    // Prefer the event's own ID (the relay's event-id header): a re-published outbox
    // row lands at a new offset. Fall back to the record's coordinates without one.
    var idempotencyKey = result.Message.Headers is { } headers
                         && headers.TryGetLastBytes("event-id", out var eventId)
        ? Encoding.UTF8.GetString(eventId)
        : $"{result.Topic}:{result.Partition.Value}:{result.Offset.Value}";

    if (await redis.KeyExistsAsync(idempotencyKey))
    {
        logger.LogDebug("Skipping duplicate message at {Key}", idempotencyKey);
        return;
    }

    var order = JsonSerializer.Deserialize<OrderEvent>(result.Message.Value)
        ?? throw new JsonException("Payload was JSON null");
    await orderRepository.UpsertAsync(order);   // an upsert: repeating it is harmless

    // Record the idempotency key AFTER the downstream write succeeds
    await redis.StringSetAsync(idempotencyKey, "1", TimeSpan.FromDays(7));
}

Correct operation order:

  Consume message
       │
       ▼
  Check idempotency store ──► Already exists? ──► Skip (caller commits)
       │
       ▼ (not seen)
  Process (downstream write)
       │
       ▼
  Record idempotency key
       │
       ▼
  Commit offset

Error Routing: Bad Records, Transient Failures and Permanent Failures

A ConsumeException most often reports a per-record failure: a key or value deserialization error (ErrorCode.Local_ValueDeserialization). Its ConsumerRecord property still holds the raw key and value bytes, and the consumer's position has already moved past the record, so it is a dead-letter case, not a retry case. Transient broker and network problems, such as a broker restart or a dropped connection, are different: librdkafka retries those internally and reports them to the handler you register with SetErrorHandler; they are not thrown from Consume.

An application exception from your handler needs classifying: a transient one (a DbException whose IsTransient is true, a timeout, an HTTP 503) deserves a bounded retry; a permanent one (JsonException, a failed validation) goes to the DLT.

const int MaxAttempts = 5;   // 4 waits: 2 + 4 + 8 + 16 s

while (!stoppingToken.IsCancellationRequested)
{
    ConsumeResult<string, string> result;
    try
    {
        result = consumer.Consume(stoppingToken);
    }
    catch (ConsumeException ex) when (ex.Error.Code is ErrorCode.Local_KeyDeserialization
                                                 or ErrorCode.Local_ValueDeserialization)
    {
        // Bad bytes (a key or value deserialization failure). The raw record is on the
        // exception and the position has already moved past it: dead-letter it,
        // then commit past it explicitly.
        await RouteRawToDeadLetterAsync(ex.ConsumerRecord, ex);
        try
        {
            consumer.Commit(new[] { new TopicPartitionOffset(
                ex.ConsumerRecord.TopicPartition, ex.ConsumerRecord.Offset + 1) });
        }
        catch (KafkaException commitEx)   // logged, like the commit at the end of the loop
        {
            logger.LogWarning(commitEx, "Commit failed at {Position}",
                ex.ConsumerRecord.TopicPartitionOffset);
        }
        continue;
    }
    catch (ConsumeException ex) when (ex.Error.Code == ErrorCode.Local_MaxPollExceeded)
    {
        // The loop overran max.poll.interval.ms and left the group; the next Consume rejoins.
        logger.LogWarning(ex, "Poll interval exceeded; rejoining the group");
        continue;
    }

    var attempt = 0;
    while (true)
    {
        try
        {
            attempt++;
            await ProcessWithIdempotencyAsync(result, redis);
            break;                                          // success
        }
        catch (Exception ex) when (!stoppingToken.IsCancellationRequested
                                   && IsTransient(ex) && attempt < MaxAttempts)
        {
            // Bounded in-process retry: 2 + 4 + 8 + 16 = 30 s of waiting, plus the
            // attempts' own run time. Give each attempt a timeout (and let IsTransient
            // count it as transient) so the total stays well inside max.poll.interval.ms
            // (300 s by default).
            logger.LogWarning(ex, "Transient failure {Attempt}/{Max} at {Position}",
                attempt, MaxAttempts, result.TopicPartitionOffset);
            await Task.Delay(TimeSpan.FromSeconds(Math.Pow(2, attempt)), stoppingToken);
        }
        catch (Exception ex) when (!stoppingToken.IsCancellationRequested)
        {
            // Permanent failure, or transient attempts exhausted (a timed-out attempt
            // included): quarantine it. Throws if the DLT write fails, so the commit
            // below is never reached.
            await RouteToDeadLetterAsync(result, ex, attempt);
            break;
        }
    }

    try
    {
        consumer.Commit(result);   // reached only after success or an acknowledged DLT write
    }
    catch (KafkaException ex)
    {
        // Typically the partition moved in a rebalance; its next owner re-reads the message.
        logger.LogWarning(ex, "Commit failed at {Position}", result.TopicPartitionOffset);
    }
}

Three things make this loop safe. Each commit sits outside the handler's try, in a try of its own, so a commit failure during a rebalance is never mistaken for an application error and dead-lettered. It is only logged: the partition's next owner re-reads the message, and the idempotency check skips it if it was already handled. The retry is bounded and in-process, so the message is retried rather than silently passed over. And the retry and DLT catches test stoppingToken, not the exception type: an exception during shutdown ends the loop with the message uncommitted, so shutting down never dead-letters a healthy message, while a timed-out attempt counts as a failure even when it throws OperationCanceledException. Local_MaxPollExceeded means the loop overran max.poll.interval.ms and left the group: it is logged, and the next Consume rejoins. Any other ConsumeException, such as an authorization failure or an unknown topic, is usually a configuration problem and is left to stop the service. The retry pauses every partition this instance owns; if a dependency can be down for minutes, stop consuming (or Pause the partitions) instead of retrying longer.

Dead-Letter Routing: Order of Operations Under Failure

The ordering constraint is strict: write to the DLT, await its acknowledgement, then commit the original offset. Commit first and crash before the DLT write completes, and the message is never processed and never inspected. RouteToDeadLetterAsync from the dead-letter section does the awaiting; the loop above does the committing. Records that failed to deserialize need a byte-level twin, because there is no typed value to forward, and the replay side reads the same text headers back:

async Task RouteRawToDeadLetterAsync(ConsumeResult<byte[], byte[]> failed, Exception reason)
{
    await _rawDltProducer.ProduceAsync($"{failed.Topic}.DLT", new Message<byte[], byte[]>
    {
        Key = failed.Message.Key,
        Value = failed.Message.Value,
        Headers = DeadLetterHeaders(failed.Message.Headers, failed.TopicPartitionOffset, reason, attempts: 0)
    });
}

// The replay side reads the provenance back from the same text headers.
static TopicPartitionOffset OriginalPosition(Headers h) => new(
    Encoding.UTF8.GetString(h.GetLastBytes("dlt.original.topic")),
    new Partition(int.Parse(Encoding.UTF8.GetString(h.GetLastBytes("dlt.original.partition")), CultureInfo.InvariantCulture)),
    new Offset(long.Parse(Encoding.UTF8.GetString(h.GetLastBytes("dlt.original.offset")), CultureInfo.InvariantCulture)));

Every path uses the same <topic>.DLT name and the same UTF-8 text headers. Topic names are case-sensitive, and binary header encodings such as BitConverter are platform-dependent, so mixing conventions quietly splits your dead letters across topics and formats.

The Outbox Relay Worker in .NET

The outbox pattern solves the dual-write problem on the producer side. The .NET implementation is a BackgroundService that continuously polls an outbox table and relays pending rows to Kafka. The critical constraint: delete the outbox row only after the broker has acknowledged the produce.

public sealed class OutboxRelayWorker : BackgroundService
{
    private readonly IServiceScopeFactory _scopeFactory;
    private readonly IProducer<string, string> _producer;
    private readonly ILogger<OutboxRelayWorker> _logger;

    public OutboxRelayWorker(IServiceScopeFactory scopeFactory, ILogger<OutboxRelayWorker> logger)
    {
        _scopeFactory = scopeFactory;
        _logger = logger;

        var producerConfig = new ProducerConfig
        {
            BootstrapServers = "localhost:9092",
            EnableIdempotence = true,   // no duplicates/reordering from the client's OWN retries
            Acks = Acks.All
        };
        _producer = new ProducerBuilder<string, string>(producerConfig).Build();
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        while (!stoppingToken.IsCancellationRequested)
        {
            using var scope = _scopeFactory.CreateScope();
            var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();

            // Assumes ONE relay instance. With several replicas, claim rows first
            // (e.g. PostgreSQL SELECT ... FOR UPDATE SKIP LOCKED, or a leader lease),
            // or two relays will publish the same row.
            var pendingRows = await db.OutboxMessages
                .OrderBy(m => m.Sequence)   // a DB-assigned identity column, not an app clock
                .Take(50)
                .ToListAsync(stoppingToken);

            var failed = false;
            foreach (var row in pendingRows)
            {
                try
                {
                    await _producer.ProduceAsync(row.Topic, new Message<string, string>
                    {
                        Key = row.Key,
                        Value = row.Payload,
                        // Consumers deduplicate on this ID: a re-published row lands at a
                        // NEW offset, so topic:partition:offset cannot recognise it.
                        Headers = new Headers { { "event-id", Encoding.UTF8.GetBytes(row.Id.ToString()) } }
                    }, stoppingToken);
                }
                catch (ProduceException<string, string> ex)
                {
                    // Stop this batch: publishing later rows first would reorder events
                    // for the same key. The failed row is retried on the next poll.
                    _logger.LogWarning(ex, "Produce failed for outbox row {Id}", row.Id);
                    failed = true;
                    break;
                }

                // Delete only after the broker has acknowledged the write.
                db.OutboxMessages.Remove(row);
                await db.SaveChangesAsync(stoppingToken);
            }

            if (failed)
                await Task.Delay(TimeSpan.FromSeconds(5), stoppingToken);   // don't spin on a failing row
            else if (pendingRows.Count == 0)
                await Task.Delay(TimeSpan.FromMilliseconds(500), stoppingToken);
        }
    }

    public override void Dispose()
    {
        _producer.Flush(TimeSpan.FromSeconds(5));   // don't drop anything still queued
        _producer.Dispose();
        base.Dispose();
    }
}

Three details matter as much as the ordering. On a failed produce the loop stops the batch instead of moving on: continuing would publish row N+1 before row N's retry and reorder events for the same key. Rows are read in the order of a database-assigned identity column, not an app-server clock like CreatedAt: clocks on different instances drift, and a later event could sort first. (Per-key order also assumes that transactions writing the same key run one after another, for example under a row lock.) And every record carries the row's ID as an event-id header. EnableIdempotence = true only prevents duplicates from the client's own internal retries; when this worker publishes a row again after a timeout or a restart, the broker sees a brand-new record, and it is the consumer's dedupe on event-id that makes the second copy harmless.

Putting the Pieces Together: The Full Flow

Producer side (Outbox Worker):

  Business operation
  + DB write
  + Outbox row write  ──► Single DB transaction (atomic)
         │
         ▼
  Poll outbox rows
         │
         ▼
  ProduceAsync ──► Broker ACK
         │
         ▼
  Delete outbox row

Consumer side (BackgroundService):

  Consume message
         │
         ├── ConsumeException (bad bytes) ──► raw record to DLT (await ack)
         │                                    Commit past it
         ▼
  Check idempotency store
         ├── Seen ──► Commit & skip
         │
         ▼ (unseen)
  Process message
         │
         ├── Transient exception ──► bounded in-process retry
         ├── Permanent / retries exhausted ──► DLT (await ack)
         │                                     Commit original offset
         ▼
  Record idempotency key
         │
         ▼
  Commit offset

The ordering at every handoff — produce before delete, DLT before commit, idempotency key after downstream write — is what makes the full pipeline safe under process crashes, broker restarts, and rebalances.

📋 Pattern Ordering Rules

🧠 Mnemonic: "ACK before advance" — whether advancing a database row out of the outbox or advancing a Kafka offset, wait for acknowledgment from the next system before removing proof that the work needs doing.

✍️ Exercise: trace the silent skip

A consumer uses pairing (a) and handles transient errors like this — no Seek, no commit:

catch (DbException ex) when (ex.IsTransient)
{
    _logger.LogWarning(ex, "Transient DB error, will retry");
    await Task.Delay(retryDelay, cancellationToken);   // then the loop calls Consume again
}

Partition 0 holds offsets 41 and 42. Processing 41 hits a transient database error; 42 succeeds.

1. What does the next Consume return after the delay?
2. What offset does Commit(result) commit after 42 succeeds?
3. Is 41 ever processed?
4. If you use the in-process loop from "Error Routing" instead, which makes up to 5 attempts with 2, 4, 8 and 16 s waits between them, how long can one message hold the loop, and is that safe?

Check your answer

1. Offset 42. Consume had already moved the position to 42 when it returned 41.
2. 43 (Commit records offset + 1).
3. No. The group's committed offset is now past it; only a manual rewind would reach it. Not committing did not bring it back; the later commit skipped it.
4. The waits add up to 2 + 4 + 8 + 16 = 30 s (the fifth failure, a timeout included, goes to the DLT without waiting), plus the processing attempts themselves. With a short timeout on each attempt, that stays well inside the 300 s default max.poll.interval.ms, so the consumer is not evicted. But every partition this instance owns is paused for that time, so keep the retry budget short.

---

Shutdown, Tracing and the Dashboard

The patterns above keep data correct. Three more make the service operable: it must stop without stalling the group, a request must be traceable across the topic, and the dashboard must show the failure modes this lesson describes.

Graceful shutdown

When the host stops, it cancels stoppingToken and waits for every hosted service to finish, up to HostOptions.ShutdownTimeout (30 seconds by default in current .NET). Inside that budget, a Kafka service has to:

• Close the consumer. consumer.Close() leaves the group, so its partitions are reassigned immediately. Disposing without closing makes the group wait out session.timeout.ms (45 s by default) before anyone takes those partitions over. Close() commits final offsets only when auto-commit is on; with pairing (a), your last Commit(result) already did that.
• Flush the producer. Flush(timeout) sends what is still queued and returns how many messages are left. Disposing a producer without flushing drops anything queued or in flight.
• Finish or abandon the current message. Pass the token into the handler. A handler that is cancelled throws, the offset is not committed, and the next owner reprocesses the message, which is harmless with idempotent processing.

public sealed class OrderConsumerService(
    ConsumerConfig config, ILogger<OrderConsumerService> logger) : BackgroundService
{
    // Consume() blocks. Running the loop on its own thread keeps it off the host's
    // startup path on every .NET version (.NET 10 already starts ExecuteAsync in the
    // background; earlier versions ran its synchronous prefix during startup).
    protected override Task ExecuteAsync(CancellationToken stoppingToken) =>
        Task.Run(() => RunAsync(stoppingToken), stoppingToken);

    private async Task RunAsync(CancellationToken stoppingToken)
    {
        using var consumer = new ConsumerBuilder<string, string>(config).Build();
        consumer.Subscribe("orders");
        try
        {
            while (!stoppingToken.IsCancellationRequested)
            {
                var result = consumer.Consume(stoppingToken);
                // Pass the token on: a handler cancelled mid-message throws, the offset is
                // not committed, and the next owner reprocesses it (at-least-once).
                await HandleAsync(result, stoppingToken);
                consumer.Commit(result);
            }
        }
        catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
        {
            logger.LogInformation("Shutting down consumer");
        }
        finally
        {
            consumer.Close();   // leave the group now, so partitions move immediately
        }
    }

    private Task HandleAsync(ConsumeResult<string, string> r, CancellationToken ct) => Task.CompletedTask;
}

Keep the host's budget below the platform's: Kubernetes sends SIGTERM and waits terminationGracePeriodSeconds (30 s by default) before killing the pod, so a 30 s host timeout leaves no margin. Set it explicitly, for example builder.Services.Configure<HostOptions>(o => o.ShutdownTimeout = TimeSpan.FromSeconds(25));.

Carrying a trace across Kafka

A message crosses process boundaries, so the trace (one request's tree of timed operations, called spans) has to travel with it, in the record's headers. The W3C Trace Context standard defines a traceparent header: 00-{32 hex trace-id}-{16 hex parent span id}-{2 hex flags}. The producer writes the current context into the headers; the consumer reads it back and starts its processing span as a child, so one trace shows the HTTP request, the send, and the process.

public static class KafkaTracing
{
    public static readonly ActivitySource Source = new("Orders.Messaging");

    // W3C trace context + W3C baggage (app key-value pairs that travel with the trace),
    // set explicitly so it works without SDK defaults.
    private static readonly TextMapPropagator Propagator = new CompositeTextMapPropagator(
        new TextMapPropagator[] { new TraceContextPropagator(), new BaggagePropagator() });

    // Producer side: start a PRODUCER activity and write its context into the headers.
    public static Activity? StartSend(string topic, Headers headers)
    {
        var activity = Source.StartActivity($"send {topic}", ActivityKind.Producer);
        activity?.SetTag("messaging.system", "kafka");
        activity?.SetTag("messaging.destination.name", topic);
        activity?.SetTag("messaging.operation.name", "send");
        activity?.SetTag("messaging.operation.type", "send");

        var context = activity?.Context ?? Activity.Current?.Context ?? default;
        Propagator.Inject(new PropagationContext(context, Baggage.Current), headers,
            (h, key, value) => { h.Remove(key); h.Add(key, Encoding.UTF8.GetBytes(value)); });
        return activity;
    }

    // Consumer side: continue the producer's trace as the parent of the processing span.
    public static Activity? StartProcess(ConsumeResult<string, string> result)
    {
        var parent = Propagator.Extract(default, result.Message.Headers,
            (h, key) => h is not null && h.TryGetLastBytes(key, out var bytes)
                ? new[] { Encoding.UTF8.GetString(bytes) }
                : Array.Empty<string>());
        Baggage.Current = parent.Baggage;

        // The conventions ask for a link to the message's context: a reference to another span
        // that, unlike a parent, does not nest this span under it (a span can have many links).
        // Making it the parent as well is optional.
        var links = parent.ActivityContext == default
            ? null
            : new[] { new ActivityLink(parent.ActivityContext) };
        var activity = Source.StartActivity(
            $"process {result.Topic}", ActivityKind.Consumer, parent.ActivityContext, links: links);
        activity?.SetTag("messaging.system", "kafka");
        activity?.SetTag("messaging.destination.name", result.Topic);
        activity?.SetTag("messaging.operation.name", "process");
        activity?.SetTag("messaging.operation.type", "process");
        return activity;
    }
}

The span names (send orders, process orders), the PRODUCER and CONSUMER kinds, the messaging.system, messaging.destination.name and messaging.operation.* attributes, and the processing span's link to the message's context follow the OpenTelemetry messaging semantic conventions. Those conventions are still evolving, so pin the version you follow. The OpenTelemetry .NET contrib project also ships a Confluent.Kafka instrumentation package; check its release status before depending on it. Using the message's context as the parent as well, as this code does, is allowed only for a single message: with a batch consumer, one processing span covers many messages and a span can have only one parent, so link to each message's context instead.

A dashboard that tells the truth

Each panel answers a failure mode from this lesson:

• Per-partition lag, from an exporter that does not run inside the consumer; show the sum only as an overview.
• Committed-offset rate per partition. A rate of zero on one partition while its log-end offset keeps rising and the other partitions keep committing is the poison-pill signature.
• Time to catch up, from the net drain rate (consume − produce); rising means the backlog is clearing more slowly; empty while lag grows means it is not clearing at all.
• Group state and member count (DescribeConsumerGroupsAsync). Empty or Dead means nobody is consuming, whatever lag says.
• Rebalances. librdkafka's statistics report a rebalance_cnt per consumer instance; frequent rebalances with no deploy usually mean members are being evicted (max.poll.interval.ms or session.timeout.ms) or restarting.
• DLT write rate. Anything above zero deserves a look.
• Processing latency and errors, from the process spans.

librdkafka can also emit a statistics JSON document per client: set StatisticsIntervalMs (it is 0, off, by default) and register SetStatisticsHandler. Per partition it includes committed_offset, hi_offset, ls_offset and consumer_lag, which is the high watermark (or the last stable offset under read_committed) minus the committed offset. The high watermark is the offset just past the last fully replicated message, the number lag tools report as the log-end offset. It is useful detail, but it comes from the consumer itself, so it vanishes when the consumer does. Keep the external exporter as the primary lag signal.

✍️ Exercise: does the shutdown fit?

A service runs with the default 30 s ShutdownTimeout in a pod with the default 30 s grace period. Its handler can take up to 20 s, and it ignores the cancellation token. Shutdown flushes its producer with a 15 s bound, then closes the consumer. What can go wrong, and what do you change?

Check your answer

20 s (handler) + up to 15 s (flush) = 35 s, more than the 30 s host budget, so the host can give up before Close() runs. The pod's own 30 s grace period ends at about the same moment, and the process is killed. Without Close(), the group waits out session.timeout.ms (45 s) before reassigning the partitions. And if the kill lands during the flush, queued messages are dropped.

Fixes:

• Pass stoppingToken into the handler, so an in-flight message is abandoned and reprocessed rather than finished.
• Bound the flush to a few seconds.
• Set ShutdownTimeout comfortably below the pod's grace period, for example 25 s against 30 s, or raise the grace period.

---

Common Mistakes and What They Cost

Every mistake covered here has the same underlying shape: a gap between what the code appears to guarantee and what it actually guarantees under failure. Locally, these gaps are invisible — the happy path never exercises them. In production, they surface as silent data loss, phantom consistency, or monitoring that lies about system health.

Mistake 1: Committing the Offset Before the Downstream Write Completes

Premature offset commitment is the single most consequential sequencing error in consumer code, and it is easy to introduce accidentally — especially when enable.auto.commit=true is left in place or when a Commit() call is placed at the top of a retry block.

Message received (offset 47)
        │
        ▼
  consumer.Commit(result)  ◄── group's committed offset is now 48 (next to read)
        │
        ▼
  downstream write   ◄── CRASH HERE
        │
        ▼
  (never completes)

After the crash, the consumer group restarts from offset 48. The group will not read 47 again — it is still in the log, but only a manual rewind would reach it. If the downstream write was an order fulfillment record or a payment event, that event simply did not happen, with nothing in your system flagging it as missing.

The correct ordering is: receive → process → write downstream → commit. With auto-commit, the offset is stored the moment Consume hands you the message, and the timer (auto.commit.interval.ms, default 5000 ms) ticks on its own schedule. If the write is slow and the process dies 4 seconds after receiving the message, any tick in those 4 seconds has already committed past it — with a 5-second interval, four times out of five. EnableAutoCommit = false with a commit after the write removes this ambiguity entirely.

Mistake 2: Swallowing All Exceptions and Always Committing

This mistake usually starts with good intentions: keep the consumer moving so it doesn't fall behind.

// ❌ DO NOT DO THIS — silent data loss pattern
while (!stoppingToken.IsCancellationRequested)
{
    try
    {
        var result = consumer.Consume(stoppingToken);
        await ProcessMessageAsync(result.Message);
    }
    catch (Exception ex)
    {
        _logger.LogError(ex, "Error processing message, skipping.");
    }
    finally
    {
        consumer.Commit(); // always commit so we don't reprocess
    }
}

This code has three compounding problems. The commit always fires even when ProcessMessageAsync threw because the downstream API was temporarily unavailable — that message is now permanently gone. The DLT becomes unreachable because failed messages are never routed there before committing. And transient errors and poison pills look identical in the logs, destroying the signal that distinguishes a bad message from a bad environment.

🎯 Key Principle: A consumer that never stops but also never correctly processes is not more reliable than one that stops — it is less reliable, because it produces no error signal and silently discards work.

Mistake 3: Not Testing Consumer Behavior Under Partition Rebalance

A rebalance occurs when consumers join or leave a group, or when a consumer fails to poll within max.poll.interval.ms. During a rebalance, partition assignments are revoked and redistributed. Consider a consumer whose poll loop keeps calling Consume, buffering messages in memory for a worker to process, and which commits once when all 500 of a batch are done:

Consumer A owns partition 3
  - Reads messages 100–599
  - Processes messages 100–299
  - Rebalance triggered (another instance joined)
  - Partition 3 revoked from Consumer A — NO COMMIT YET
  - Consumer B assigned partition 3, resumes from offset 100
  - Messages 100–299 processed again ✓ (if idempotent) or duplicated ✗

(The revoked handler runs inside Consume (or Close), so a loop that stops polling while it works is never revoked mid-batch; if it overruns max.poll.interval.ms, its partitions are lost instead.)

The practical fix registers a rebalance handler that commits the work actually finished on revocation — here, offset 300 — so Consumer B resumes at 300 instead of 100:

// Next offset to commit for each partition, recorded only AFTER a message is fully
// processed:  processed[result.TopicPartition] = result.Offset + 1;
var processed = new ConcurrentDictionary<TopicPartition, Offset>();

var consumer = new ConsumerBuilder<string, string>(config)   // EnableAutoCommit = false
    .SetPartitionsRevokedHandler((c, revoked) =>
    {
        // Commit only finished work. A parameterless c.Commit() would commit the
        // client's STORED offsets — by default, everything Consume has handed out,
        // including messages still waiting unprocessed in memory.
        var done = revoked
            .Where(tpo => processed.ContainsKey(tpo.TopicPartition))
            .Select(tpo => new TopicPartitionOffset(tpo.TopicPartition, processed[tpo.TopicPartition]))
            .ToList();
        if (done.Count > 0) c.Commit(done);
        foreach (var tpo in revoked) processed.TryRemove(tpo.TopicPartition, out _);
    })
    .SetPartitionsLostHandler((c, lost) =>
    {
        // Lost, not revoked: another member may already own these partitions, so a
        // commit would fail. Drop the bookkeeping; the new owner replays the gap.
        foreach (var tpo in lost) processed.TryRemove(tpo.TopicPartition, out _);
    })
    .Build();

With a worker thread, both handlers must also make the worker let go of those partitions: drop their buffered messages and, before the revoked handler commits, wait for the one in flight. Otherwise A's worker goes on processing 300–599 while B processes them too.

Don't reach for a parameterless c.Commit() here. It commits the client's stored offsets, and with the default automatic offset store those include every message Consume has handed out — in this scenario, 600, which would skip 300–599 entirely. (With auto-commit on, librdkafka commits the stored offsets on revocation by itself, which carries the same risk unless you store offsets only after processing.)

⚠️ Common Mistake: Relying on integration tests that run a single consumer instance against a single-partition topic. A lone member does rebalance — when it joins, when its subscription changes, when partitions are added, or when it exceeds max.poll.interval.ms — but never because a peer joined mid-batch, which is the case that exposes the bug above. Test with at least two instances and a partition count that allows redistribution.

Mistake 4: Treating Zero Lag as Proof the Consumer Is Healthy

This mistake lives in the monitoring layer, but its consequences are just as severe.

Scenario A — healthy consumer, busy topic:
  Log-end offset:           1,000
  Committed offset:         1,000
  Lag:                          0  ✓ Consumer is keeping up

Scenario B — consumer group died at 02:00; the topic is quiet overnight:
  Log-end offset:           1,000
  Committed offset:         1,000  (frozen since 02:00)
  Lag:                          0  ✗ Nobody is consuming

  [07:30 — the first morning orders arrive, offsets 1,000 ... 1,049]
  Log-end offset:           1,050
  Committed offset:         1,000  (still frozen)
  Lag:                         50  ← only now does lag move

Lag measures backlog, not liveness. On a quiet topic a dead group shows zero lag for as long as the topic stays quiet — five and a half hours here — and the first sign of trouble is the morning's orders not being processed. The same blind spot appears when lag is emitted by the consumer process itself: when it dies, the series stops instead of rising, and many dashboards draw that as a flat line. The additional signal you need is consumer group liveness: DescribeConsumerGroupsAsync returns each group's state and members. Alert when lag is growing or the group has no members / is not Stable for longer than a normal rebalance — every deploy passes briefly through PreparingRebalance and CompletingRebalance, so alerting on the state alone pages you on every release.

Mistake 5: Skipping the Outbox Pattern and Producing Inside a Database Transaction

This introduces a distributed consistency gap that is invisible in local development and intermittent in production.

// ❌ Dual-write anti-pattern
public async Task PlaceOrderAsync(Order order)
{
    using var transaction = await _db.Database.BeginTransactionAsync();

    _db.Orders.Add(order);
    await _db.SaveChangesAsync();

    // ProduceAsync runs BEFORE the DB commit. If it succeeds and the commit then
    // fails (or the process dies here), the event is in Kafka but the order is not.
    await _producer.ProduceAsync("orders", new Message<string, string>
    {
        Key = order.Id.ToString(),
        Value = JsonSerializer.Serialize(order)
    });

    await transaction.CommitAsync();
    // Swap the two steps and you get the mirror image: order committed, event never sent.
}

This gap cannot be closed by wrapping the produce in the same database transaction, because Kafka and relational databases do not share a transaction coordinator. The correct solution is the outbox pattern shown in "The Outbox Relay Worker in .NET" above: write the event to a database table in the same transaction as the business write, then have a separate relay worker produce to Kafka, deleting the row only after receiving a successful broker acknowledgment.

// ✅ Outbox pattern — event write is atomic with the business write
public async Task PlaceOrderAsync(Order order)
{
    using var transaction = await _db.Database.BeginTransactionAsync();

    _db.Orders.Add(order);

    _db.OutboxMessages.Add(new OutboxMessage
    {
        Id = Guid.NewGuid(),
        Topic = "orders",
        Key = order.Id.ToString(),
        Payload = JsonSerializer.Serialize(order),
        CreatedAt = DateTime.UtcNow
    });

    await _db.SaveChangesAsync();
    await transaction.CommitAsync();
    // Both Order and OutboxMessage are committed atomically.
    // The relay BackgroundService will produce OutboxMessages to Kafka.
}

✍️ Exercise: review this billing consumer

var consumer = new ConsumerBuilder<string, string>(new ConsumerConfig
{
    BootstrapServers = "kafka:9092",
    GroupId = "billing"               // every other setting left at its default
}).Build();
consumer.Subscribe("invoices");

while (!ct.IsCancellationRequested)
{
    var r = consumer.Consume(ct);
    try { await ChargeCustomerAsync(r.Message.Value); }
    catch (Exception ex) { _log.LogError(ex, "Charge failed"); }
}
// Alerting: a gauge inside this same service reports lag; the alert fires at lag > 10,000.

Name three ways this service can lose charges or hide the loss.

Check your answer

1. Auto-commit with the automatic offset store (both defaults): each offset is stored the moment Consume returns the message and committed on the 5-second timer. A crash mid-charge can leave the offset committed past a charge that never happened (Mistake 1).
2. Catch, log, and move on (Mistake 2): a failed charge is neither retried nor dead-lettered, and the next commit carries the position past it. The charge is gone, with only a log line to show for it.
3. Lag measured inside the service, alerted on a raw count (Mistake 4 and the time-to-catch-up section): when the service dies, its gauge dies with it, so lag never crosses 10,000 on the dashboard. And even while it is alive, 10,000 messages means seconds on a busy topic and hours on a quiet one.

Fixes, in the lesson's terms:

• Use pairing (a) and commit after the charge.
• Retry transient failures in a bounded loop, and send permanent ones to the DLT before committing.
• Deduplicate charges on an event ID.
• Export lag from a separate service, alert on time to catch up and on group state, and close the consumer on shutdown.

---

Summary

Production Kafka demands that you reason about failure at the pipeline level, design for delivery semantics deliberately, and distinguish between failure types before deciding how to respond. The patterns in this lesson are the concrete application of that mindset.

📋 The Reliability Stack

📋 The Five Costly Mistakes

None of these five mistakes produces an error log or a metric alert under normal conditions. They are only visible when specific failure modes are exercised — crashes, broker unavailability, partition rebalances. Testing for them requires deliberately inducing those conditions.

Practical next steps:

🎯 Audit your offset commit placement. Verify every Commit() or StoreOffset() call is placed after the downstream write is acknowledged — not before, not in a finally block, not left to the default automatic offset store.

🎯 Add consumer group state to your lag dashboard. Lag-only alerting has a blind spot for dead groups on quiet topics, and for lag emitted by the consumer itself. Include group state and member count alongside lag, and write a runbook entry for the zero-lag-but-dead-group scenario.

🎯 Introduce a chaos step in your CI pipeline. Run consumer integration tests with a mid-test broker restart or a forced rebalance. If the test shows lost messages, or duplicates your idempotency layer does not catch, fix the consumer and keep the test.