Consumer & Manual Commits

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

• choose GroupId, AutoOffsetReset, EnableAutoCommit and EnableAutoOffsetStore deliberately, and say what each combination can lose or duplicate;
• write a Consume(CancellationToken) loop that separates poison messages from fatal errors and shuts down cleanly with Close();
• store and commit offsets only after processing succeeds — including the off-by-one between a record's offset and the offset you commit;
• commit correctly around rebalances, and explain why a slow handler causes one;
• make a handler safe to run twice.

Prerequisites: "Core Kafka Mental Model" (offsets as positions), "Consumer Groups & Offsets" (groups, committed offsets) and "Raw Confluent.Kafka Client" (building a consumer). Each major section includes an exercise — try it before opening the answer.

📹Watch the Lesson Video29 min

Consumer Configuration for Correctness and Control

Picture a billing consumer that ran flawlessly in staging, then shipped to production and quietly skipped a full day of invoices the moment it was deployed as a brand-new consumer group. Nobody touched the processing logic. The bug was sitting in the ConsumerConfig the whole time — a handful of flags that decide, before a single line of your handler code executes, whether messages get lost, replayed, or processed exactly when you expect. This section is about those flags: GroupId, AutoOffsetReset, EnableAutoCommit, EnableAutoOffsetStore, and the timing knobs session.timeout.ms, heartbeat.interval.ms, and max.poll.interval.ms. Get these wrong and no amount of careful commit logic later will save you.

GroupId: What You're Actually Opting Into

GroupId is not a label for your consumer — it's a coordination key. Every consumer instance that connects with the same GroupId becomes a member of the same consumer group, and the group divides the subscribed topic's partitions across the currently active members (in the classic protocol an elected consumer computes the assignment and the coordinator distributes it) so that, at steady state, each partition is owned by exactly one member of the group at a time.

var consumerConfig = new ConsumerConfig
{
    BootstrapServers = "localhost:9092",
    GroupId = "billing-service", // members sharing this string share partitions
    AutoOffsetReset = AutoOffsetReset.Earliest,
    EnableAutoCommit = false,
    EnableAutoOffsetStore = false
};

The practical consequence: if you run three instances of billing-service with the same GroupId against a topic with six partitions, each instance ends up owning roughly two partitions — and Kafka, not your code, decides which ones. If you instead give each instance a different GroupId, they stop cooperating entirely; each one behaves as an independent group and receives all partitions, meaning every instance reads every message. This is the classic mistake behind "why is my downstream system getting triplicate events" — someone generated a unique GroupId per pod (for example, embedding a hostname or GUID), and each pod ended up as a group of one, happily consuming the entire topic on its own.

🎯 Key Principle: GroupId answers the question "who am I sharing partitions with?" — not "what should I call myself in logs." Two consumers with the same GroupId divide the work; two consumers with different GroupIds duplicate the work.

The mechanics of how partitions get reassigned when a member joins or leaves — the rebalance protocol itself — belongs to a different part of this lesson, in "Partition Assignment, Rebalances, and Commit Timing." What matters here is simpler: GroupId is the first configuration decision you make, and it determines whether your fleet of consumer instances behaves as one cooperating group or as several independent, fully-duplicating readers.

AutoOffsetReset: What Happens With No History

Every consumer group has committed offsets stored per topic-partition — its bookmark. But a brand-new consumer group has no bookmark at all, and AutoOffsetReset decides what happens the first time that group tries to read a partition with nothing committed.

The distinction only matters when there's no committed offset to resume from — once a group has committed at least once for a partition, AutoOffsetReset is irrelevant for that partition; the consumer resumes from the committed offset every time, regardless of this setting. It only fires on first contact, or when the committed offset is no longer usable — either the group's committed offsets expired (the broker drops them after offsets.retention.minutes once the group has been empty that long), or the log segments holding that offset have already been deleted by the topic's retention.

In Confluent.Kafka, AutoOffsetReset defaults to Latest if you don't set it — meaning an unconfigured new consumer group in production will silently start from "whatever comes in from now on" and skip the entire existing backlog. For a service like billing-service, that default is exactly how you lose a full day of invoices: the group is new, the topic already has a full day of orders sitting in it, and Latest tells the consumer "none of that concerns you." Setting AutoOffsetReset = AutoOffsetReset.Earliest explicitly, as in the config above, is the safer default for most business-critical consumers — you'd rather reprocess a backlog you've already handled (which idempotent processing absorbs) than silently skip data you haven't.

💡 Pro Tip: Treat AutoOffsetReset as a decision you write down, not a default you inherit. If you catch yourself relying on whatever the library ships with, you've already made the decision by accident.

EnableAutoCommit and EnableAutoOffsetStore: Two Switches, Not One

This pair causes more confusion than any other setting in ConsumerConfig, because it looks like one decision ("manual commits, yes or no?") but is actually two independent mechanisms layered on top of each other.

Internally, Confluent.Kafka's consumer keeps a local, in-memory offset store per assigned partition — think of it as "the offset I'd commit next if asked." Two separate things touch that store:

🔧 EnableAutoOffsetStore controls whether the client automatically writes an offset into that in-memory store every time Consume() hands you a message. Specifically, the offset after that message (offset + 1, the position to resume from) is stored automatically, before you've done anything with it.

🔧 EnableAutoCommit controls whether a background timer periodically takes whatever is currently sitting in that in-memory store and commits it to the broker, on a fixed interval, independent of whether you've finished processing anything.

The two flags default to true in Confluent.Kafka. That combination — auto-store plus auto-commit — is what makes the library usable out of the box for a demo: consume a message, the offset gets stored immediately, and a timer commits it later, no matter what your code did with the message in between.

The trap is disabling only EnableAutoCommit. If you set EnableAutoCommit = false but leave EnableAutoOffsetStore at its default of true, you've stopped the periodic commit, but the client is still auto-storing the offset of every message the instant Consume() returns it — meaning that whenever you do eventually call Commit() yourself (say, on shutdown, or via some other codepath), it commits whatever was auto-stored, which may include messages you haven't actually finished processing yet. You think you have manual control; you actually have manual timing over an offset value you didn't manually set.

// The combination that gives you real manual control over WHEN and WHICH
// offset gets committed:
var consumerConfig = new ConsumerConfig
{
    BootstrapServers = "localhost:9092",
    GroupId = "billing-service",
    AutoOffsetReset = AutoOffsetReset.Earliest,
    EnableAutoCommit = false,      // no background commit timer
    EnableAutoOffsetStore = false  // no automatic "offset seen" bookkeeping either
};

// Later, in the consume loop, after processing succeeds:
consumer.StoreOffset(consumeResult); // marks the offset as safe to commit
consumer.Commit();                   // flushes the stored offset to the broker

With both flags off, StoreOffset and Commit become entirely your responsibility, called only after your handler has actually finished successfully. The full call sequence — storing after every success, committing in batches, and committing on revoke and shutdown — is assembled in "Putting It Together: The Complete Manual-Commit Loop" later in this lesson. What matters here is that EnableAutoOffsetStore is the flag deciding which offset gets committed, so it is the one that must be off for a StoreOffset/Commit() sequence to mean what you think it means. Turning off only EnableAutoCommit is the half-measure that bites: it still lets the client silently mark messages as "seen" before you've confirmed they were handled.

Two practical notes on that table. StoreOffset only works in the bottom two rows: while EnableAutoOffsetStore is true, librdkafka refuses manual stores and Confluent.Kafka throws a KafkaException. And the third row is not a compromise — it gives the same at-least-once guarantee as the fourth, with the timer doing the committing; it simply replays up to one commit interval (5 s by default) of finished work after a crash.

⚠️ Common Mistake: Setting EnableAutoCommit = false and assuming that alone gives you manual commits. It doesn't — it only removes the timer. The offset-store auto-population from EnableAutoOffsetStore keeps running unless you turn it off too. This exact gap is revisited as a standalone pitfall in "Common Pitfalls and Pre-Production Checklist," because it's one of the most repeated configuration bugs in raw Confluent.Kafka consumers.

The Timing Knobs That Decide Whether You're "Alive"

Even with commit behavior nailed down, three timing settings decide something orthogonal but equally important: whether the broker still considers your consumer a live member of the group at all. Get these wrong and you can be actively, correctly processing messages while the group has already dropped you and handed your partitions to someone else.

These describe the classic group protocol, which librdkafka still uses by default (its documentation says that default will change, so check group.protocol in your version). With the newer protocol from KIP-848 (GroupProtocol = GroupProtocol.Consumer, Kafka 4.0+ brokers), the session timeout and heartbeat interval are configured on the broker for the group instead of in ConsumerConfig, and the broker computes assignments; max.poll.interval.ms still applies on the client.

The first two operate on their own background thread inside the client and are relatively forgiving: as long as your process is alive and the network is healthy, heartbeats go out every heartbeat.interval.ms, and the broker only declares the member gone if it hears nothing for session.timeout.ms. This mechanism catches genuine failures — a crashed process, a frozen container, a severed network link.

max.poll.interval.ms is the one that catches you off guard, because it's not about whether your process is alive — it's about whether you're calling Consume() frequently enough. The client sends heartbeats from a separate thread, so a slow synchronous handler doesn't stop the heartbeats directly. But librdkafka itself tracks the time between polls, and if you take longer than max.poll.interval.ms to loop back and call Consume() again — because you're stuck doing a slow database write, a slow external HTTP call, or heavy synchronous processing on the message you just received — the client concludes you've stalled and sends a "leave group" on your behalf, which triggers a rebalance even though your process never crashed and your heartbeats never stopped. The check itself runs in the client: the broker doesn't watch your Consume() calls (under the classic protocol it only receives the value as the time it allows members to rejoin during a rebalance).

🤔 Did you know? This means a consumer can be fully "alive" by every heartbeat measure and still drop out of the group, purely because its handler is slow. The interaction between this specific timeout and slow processing — including how it produces an unwanted rebalance mid-work — is developed further in "Partition Assignment, Rebalances, and Commit Timing," since that's where the rebalance-handling mechanics live.

For now, the configuration-level takeaway is: these three numbers are not independent tuning knobs to leave at their defaults and forget. If your handler occasionally does slow work — a retry with backoff, a downstream call with a generous timeout — you need max.poll.interval.ms set high enough to cover the realistic worst case of a single poll-to-poll cycle, not the average case. Setting it too low for a legitimately slow (but not stuck) handler causes exactly the kind of self-inflicted rebalance churn that looks like a Kafka reliability problem but is actually a configuration mismatch with your own processing time.

The Ceiling You Set Before Writing Any Commit Logic

Here's the part that ties this whole section together: the combination of EnableAutoCommit, EnableAutoOffsetStore, and AutoOffsetReset doesn't just affect convenience — it sets a ceiling on which delivery semantic is even reachable, before you write a single line of processing or commit code.

If EnableAutoCommit stays true while EnableAutoOffsetStore is also left at its default, the broker will periodically receive commits on a timer, detached from whether your handler succeeded. That configuration alone makes loss possible: the timer can fire and commit an offset for a message whose processing crashed, hung, or threw halfway through, and that message is now permanently behind the committed offset, never to be redelivered. It makes duplicates possible too: a crash after processing but before the next tick replays everything since the last commit. So it gives you neither at-most-once nor at-least-once — it can do both kinds of damage, depending on where the crash lands relative to the timer. No amount of careful try/catch logic in your handler changes this, because the commit happens on the timer's schedule, not yours.

Set EnableAutoOffsetStore = false — with EnableAutoCommit = false as in this lesson's examples, or with the timer left on as in the third row of the table above — and you've removed the ceiling: you've made at-least-once delivery possible, because now no offset becomes committable until your code stores it. But turning both flags off doesn't automatically give you at-least-once; it only makes it achievable. Whether you actually get there depends on where in your code you call StoreOffset and Commit relative to your processing logic — commit before finishing the work and you're back to at-most-once risk with extra steps; commit only after the work succeeds and you get the safer default. Walking through both of those crash scenarios concretely — what a process crash mid-processing actually does to redelivery in each case — is the focus of "From Auto-Commit to At-Least-Once: Seeing the Guarantees Play Out."

✅ Correct thinking: Configuration decides what's possible; the placement of your commit calls decides what actually happens. A ConsumerConfig with manual commits enabled but a Commit() call placed before your processing logic gives you the same loss risk as leaving auto-commit on — you've just moved the risk into your own code instead of the library's timer.

❌ Wrong thinking: "I disabled EnableAutoCommit, so my consumer is at-least-once now." Disabling it only removes the timer. With EnableAutoOffsetStore still on, any parameterless Commit() you call still sends an offset stored the moment a message was handed to you, finished or not; and even with both flags off, at-least-once still requires calling StoreOffset/Commit in the right place, after successful processing. The exact call sequence is assembled in "Putting It Together: The Complete Manual-Commit Loop".

📋 Quick Reference Card: Configuration Ceiling

💡 Mental Model: Think of ConsumerConfig as pouring the foundation of a building, and your commit logic as the framing that goes on top. You can frame a house beautifully, but if the foundation was poured for a bungalow, you're not getting a three-story building no matter how good the framing is. EnableAutoCommit, EnableAutoOffsetStore, and AutoOffsetReset are the foundation; everything you do afterward in the consume loop operates within the limits they set.

✍️ Exercise: what can this configuration lose or duplicate?

Each consumer handles one message at a time and calls a payment gateway. The process is killed while the handler is in the middle of a call. For each configuration, can a message be lost, processed twice, or both?

1. EnableAutoCommit = true, EnableAutoOffsetStore = true (the defaults).
2. EnableAutoCommit = false, EnableAutoOffsetStore = true, and a periodic housekeeping task calls consumer.Commit() every 10 seconds.
3. EnableAutoCommit = true, EnableAutoOffsetStore = false, StoreOffset(result) called after the handler returns successfully.
4. EnableAutoCommit = false, EnableAutoOffsetStore = false, StoreOffset(result) then Commit() after the handler returns successfully.

Check your answer

1. Both. The offset is stored the moment Consume() returns, so a timer tick during the call commits past the unfinished message (lost if the charge never happened). A crash before the tick instead replays it, and everything else since the last commit (processed twice).
2. Both, for the same reason: Commit() sends whatever was auto-stored, including the message still in the handler. Disabling only the timer changed when commits happen, not which offset they carry.
3. Duplicates only. Nothing is stored until a message has succeeded, so the timer can never commit past unfinished work; up to one commit interval of finished messages is replayed after a crash.
4. Duplicates only, with a smaller replay window — normally just the message that was in flight. That remaining duplicate is why "At-Least-Once Is Not the Same as Safe" below matters.

Building the Consume Loop

Once the ConsumerConfig is set the way you want it — manual commits, manual offset storage, sensible timeouts — you still have to write the code that actually pulls messages off the broker. This is where a surprising number of raw Confluent.Kafka consumers go wrong: not because the config was wrong, but because the loop around Consume() was written like a demo instead of a service that has to survive bad messages, broker hiccups, and a Ctrl+C.

Wiring the Consumer: Subscribe vs. Assign

You build a consumer with ConsumerBuilder<TKey, TValue>, generic over the types your deserializers produce. For most business services that's something like ConsumerBuilder<string, string> (JSON payload as a string) or ConsumerBuilder<string, byte[]> if you're deserializing Protobuf yourself downstream.

using Confluent.Kafka;

var consumerConfig = new ConsumerConfig
{
    BootstrapServers = "localhost:9092",
    GroupId = "billing-service",
    AutoOffsetReset = AutoOffsetReset.Earliest,
    EnableAutoCommit = false,
    EnableAutoOffsetStore = false
};

using var consumer = new ConsumerBuilder<string, string>(consumerConfig)
    .SetErrorHandler((_, error) =>
        Console.WriteLine($"Broker error: {error.Code} - {error.Reason} (fatal: {error.IsFatal})"))
    .Build();

consumer.Subscribe("orders");

SetErrorHandler gives you visibility into broker-level problems (connection drops, metadata refresh failures) that the client is handling internally — these fire independently of the exceptions Consume() throws, and logging them here is cheap insurance against a silent, slowly-degrading consumer.

The call that matters most in this snippet is Subscribe. It tells Kafka "I want partitions from this topic, assigned to me dynamically as part of my consumer group." The group — not your code — decides which partitions this instance gets, and that assignment can change any time a rebalance happens — a topic covered in depth in "Partition Assignment, Rebalances, and Commit Timing." Subscribe is what almost every service should use: it's how Kafka gives you automatic load distribution across however many instances of billing-service you happen to be running.

The alternative is Assign, which hands the consumer an explicit, fixed list of TopicPartition (or TopicPartitionOffset) values instead of asking the group coordinator for an assignment:

consumer.Assign(new List<TopicPartition>
{
    new TopicPartition("orders", new Partition(0)),
    new TopicPartition("orders", new Partition(1))
});

With Assign, this consumer instance owns exactly those partitions, period — no group coordinator reshuffles them to another member, because Assign sidesteps the rebalance protocol entirely. You reach for Assign in narrower situations: a tool that needs to read a specific partition for debugging, a service that implements its own external partition-to-instance mapping instead of trusting Kafka's rebalance protocol, or a scenario where you deliberately want a single, unshared reader of a partition (for example, a stateful process that must never have two instances touching the same partition at once, and wants to control that itself rather than relying on group membership). For the vast majority of billing-service-style consumers, Subscribe is the default and Assign is the exception you reach for on purpose, not by accident.

The Loop: Consume(CancellationToken)

Once subscribed, the actual work is a loop around consumer.Consume(). This method is blocking — it parks the calling thread until a message arrives, a partition-EOF event (an opt-in end-of-partition notice) fires, or something goes wrong. Confluent.Kafka gives you two ways to call it: with a timeout (a TimeSpan, or milliseconds as an int), or with a CancellationToken.

The TimeSpan overload returns null when the timeout elapses with nothing to consume, which forces you to write a polling loop that keeps calling Consume(TimeSpan) over and over, checking for null, checking some separate shutdown flag, and spinning. It works, but it's an awkward fit for anything that needs to shut down promptly and cleanly.

using var cts = new CancellationTokenSource();

Console.CancelKeyPress += (_, e) =>
{
    e.Cancel = true; // don't let the runtime kill the process; let the loop exit gracefully
    cts.Cancel();
};

try
{
    while (!cts.IsCancellationRequested)
    {
        try
        {
            var result = consumer.Consume(cts.Token);

            if (result.IsPartitionEOF)
            {
                continue; // reached the end of the partition; nothing to process yet
            }

            Console.WriteLine(
                $"{result.Topic}[{result.Partition.Value}]@{result.Offset.Value} " +
                $"key={result.Message.Key} value={result.Message.Value}");

            // process the message here, then StoreOffset + Commit only after success
            // (the full sequence: "Putting It Together: The Complete Manual-Commit Loop")
        }
        catch (ConsumeException ex)
        {
            // never swallow this silently — log where it happened, then decide
            // whether to stop (fatal) or keep polling (transient). See below.
            Console.Error.WriteLine(
                $"Consume failed: {ex.Error.Reason} " +
                $"(fatal={ex.Error.IsFatal}) at {ex.ConsumerRecord?.TopicPartitionOffset}");

            if (ex.Error.IsFatal)
            {
                throw;
            }
        }
    }
}
catch (OperationCanceledException)
{
    // expected: cts was cancelled while Consume() was blocking
}
finally
{
    try { consumer.Close(); }
    catch (KafkaException ex) when (ex.Error.IsFatal) { } // a fatal client can't close; the using declaration still disposes it
}

Passing a CancellationToken to Consume means that when the token is cancelled — because your host is shutting down, or a supervisor asked the service to stop — the blocked call throws OperationCanceledException within a fraction of a second. Under the hood it is still a short polling loop: Confluent.Kafka waits in slices of dotnet.cancellation.delay.max.ms (100 ms by default, settable from 1 to 10000 in ConsumerConfig) and checks the token between slices. The difference from a hand-written Consume(TimeSpan) loop is that the library owns that loop and its sensible default, so your code has no timeout to pick and no shutdown flag to poll. It also composes naturally with IHostedService/BackgroundService patterns in ASP.NET Core, which hand you a CancellationToken for exactly this purpose.

⚠️ Common Mistake: Catching OperationCanceledException — or a blanket Exception — inside the while loop's try block instead of around the whole loop. With a while (!cts.IsCancellationRequested) condition the loop still exits on the next check, but with while (true), or a loop condition that checks a different token than the one passed to Consume, it spins: every pass calls Consume on the cancelled token and throws the same exception again. Either way the shutdown path is now hidden inside error handling. Let the cancellation exception propagate out of the loop, as in the example above, and put your cleanup (consumer.Close()) in a finally.

Reading a ConsumeResult

When Consume() returns successfully, it hands you a ConsumeResult<TKey, TValue> — not just the message payload, but everything you need to know where that message came from and whether there's actually a message there at all.

The fields that matter day to day:

IsPartitionEOF only ever becomes relevant if you've set EnablePartitionEof = true in your ConsumerConfig; by default it's off and every ConsumeResult you get back represents an actual message. When it is enabled and you hit the end of a partition, Consume() returns a ConsumeResult with IsPartitionEOF = true — and critically, Message on that result is null. This matters mechanically: if your loop's first move is result.Message.Value, checking IsPartitionEOF after that will already have thrown a NullReferenceException. The check has to come first, exactly as in the loop above, before you ever touch Message.

💡 Pro Tip: IsPartitionEOF is genuinely useful for batch-style consumers that need to know "I've drained everything currently available, time to flush a batch or checkpoint progress" rather than waiting indefinitely for the next message that might not arrive for a while.

⚠️ The off-by-one you must know: a committed offset means "the next offset to read", so after finishing the record at offset 43 you commit 44. Commit(result) and StoreOffset(result) add the 1 for you. Commit(IEnumerable<TopicPartitionOffset>) and StoreOffset(TopicPartitionOffset) use exactly the value you pass — hand them result.TopicPartitionOffset and you commit 43, so the last record is replayed after every restart.

✍️ Exercise: trace the committed offset

A consumer (both auto flags off) successfully processes offsets 41, 42 and 43 on partition 2, then restarts. For each variant, what offset is committed after 43, and which offset does the consumer read first after the restart?

1. consumer.Commit(result); after each message.
2. consumer.Commit(new[] { result.TopicPartitionOffset }); after each message.
3. consumer.StoreOffset(result); after each message, then one consumer.Commit(); after 43.
4. consumer.StoreOffset(result.TopicPartitionOffset); after each message, then one consumer.Commit(); after 43.

Check your answer

1. Commits 44; restarts at 44. Correct.
2. Commits 43; restarts at 43 — the last record is processed again on every restart.
3. Stores 42, 43, 44 in turn; Commit() sends 44; restarts at 44. Correct.
4. Stores 41, 42, 43; Commit() sends 43; restarts at 43 — the same off-by-one as variant 2, one step removed.

Variants 2 and 4 compile and look reasonable in review; only the overload tells you which offset is sent.

When Consume() Throws: Fatal vs. Transient

Not every problem behind a Kafka consumer shows up as a graceful return value — some surface as a ConsumeException thrown out of Consume() itself. The exception carries an Error object, and the single most important thing on it is Error.IsFatal.

catch (ConsumeException ex)
{
    if (ex.Error.IsFatal)
    {
        Console.WriteLine($"Fatal consumer error, stopping loop: {ex.Error.Code} - {ex.Error.Reason}");
        throw; // let it propagate; this consumer instance cannot recover
    }

    Console.WriteLine($"Non-fatal consume error, continuing: {ex.Error.Code} - {ex.Error.Reason}");
    // fall through to the next loop iteration and try again
}

Error.IsFatal = true means the underlying client has entered a state it cannot recover from on its own. For consumers, librdkafka documents one such case: static group membership fencing — a consumer with a GroupInstanceId is fenced because a newer instance started with the same instance id. (Most other problems, including authentication failures, are reported as non-fatal errors that the client keeps retrying.) When you see IsFatal, stop the loop and dispose the consumer. For the fencing case, librdkafka's guidance is to exit: this instance has been replaced, and rebuilding a consumer with the same instance id would simply fence the newer one in turn. Continuing to call Consume() on a fatally broken client gets you nothing more: the error is reported once, and the client then stops fetching.

Error.IsFatal = false covers the much more common case: something recoverable happened, and the client is still usable. Most transient broker-connectivity issues (a broker temporarily unreachable, a metadata refresh that needs to retry) are actually surfaced through the SetErrorHandler callback you wired up earlier while the client keeps retrying underneath, without ever throwing out of Consume() at all — which is exactly why that error handler is worth having. The ConsumeException path you catch directly in the loop mostly carries narrower, message-scoped problems — the biggest being deserialization failure — plus a few client events delivered through the consume path, such as ErrorCode.Local_MaxPollExceeded after the poll-interval watchdog has fired.

🎯 Key Principle: Treat Error.IsFatal as the fork in the road for your loop's error handling — true means stop, false means log and keep polling. Don't try to invent finer-grained recovery logic per error code unless you have a specific, known reason to; the fatal flag already encodes the distinction that matters operationally.

Poison Messages: Isolating Deserialization Failures

Here's the scenario that kills naive consumer loops in production: one badly-formed message lands in the topic — a producer bug, a schema mismatch, someone hand-crafting a test message with kafka-console-producer — and your deserializer throws. If that exception isn't handled distinctly from "the broker is unreachable," it can either crash the whole service or, worse, get treated the same as a fatal client error and stop a perfectly healthy consumer from ever reading the next, perfectly good message.

The good news is that Confluent.Kafka surfaces deserialization failures as a ConsumeException too, but with an important detail: the exception's own ConsumerRecord property usually still tells you exactly where the bad message lives, even though the typed Message.Value you wanted couldn't be produced.

try
{
    var result = consumer.Consume(cts.Token);
    // ... normal processing ...
}
catch (ConsumeException ex) when (
    ex.Error.Code == ErrorCode.Local_ValueDeserialization ||
    ex.Error.Code == ErrorCode.Local_KeyDeserialization)
{
    var badResult = ex.ConsumerRecord;
    Console.WriteLine(
        $"Poison message at {badResult.Topic}[{badResult.Partition.Value}]" +
        $"@{badResult.Offset.Value} failed to deserialize, sending to DLQ");

    // send the raw bytes (badResult.Message.Key/Value) to a dead-letter topic,
    // then store badResult.Offset + 1 so the skip is recorded, and continue
}
catch (ConsumeException ex)
{
    if (ex.Error.IsFatal)
    {
        throw;
    }

    Console.WriteLine($"Non-fatal consume error, continuing: {ex.Error.Reason}");
}

C# tries catch clauses in the order they're written, so the deserialization-specific handler above runs first and only falls through to the general ConsumeException handler for everything else. (The order is also enforced: putting the unfiltered catch (ConsumeException) first is a compile error, CS0160.) This separation is what lets one malformed record become a logged event routed to the dead-letter queue (DLQ) instead of a stack trace that takes down the process — the deserialization branch never checks IsFatal at all, because a bad payload on one message says nothing about whether the client connection itself is healthy.

⚠️ Common Mistake: Writing a single, undifferentiated catch (ConsumeException ex) block that logs ex.Message and moves on for everything. This technically "doesn't crash," but it also means you never distinguish a poison message (skip it, DLQ it, move on) from a fatal client failure (which needs the loop to actually stop) — you'll keep polling a client that will never deliver another message. After its one fatal error it has left the group, so the loop idles silently and the service looks healthy while consuming nothing.

💡 Real-World Example: Imagine orders normally carries JSON Order objects, but a misconfigured upstream service briefly publishes plain-text log lines to the same topic by mistake. Without the deserialization-specific catch, every one of those lines throws, and if your loop treats that the same as a fatal error, billing-service stops consuming entirely until someone notices and manually restarts it — even though every message after those bad lines is perfectly fine. With the pattern above, each bad line gets logged with its exact partition and offset, routed wherever your dead-letter strategy sends it, and the loop moves straight on to the next, valid order.

Getting this loop right — Subscribe for normal group membership, Consume(CancellationToken) for responsive shutdown, careful reading of IsPartitionEOF before touching Message, and a layered catch that separates poison messages from fatal client failures — is what turns "it works in the demo" into a consumer that survives a bad message at 2 AM without anyone paging on-call.

Partition Assignment, Rebalances, and Commit Timing

A raw Confluent.Kafka consumer doesn't just poll and process — it also has to survive the moment another member joins the group, a pod gets rescheduled, or the group coordinator decides your partitions belong to someone else. That moment is a rebalance, and it is where consumers that look correct in a demo start silently reprocessing work or dropping it in production. The mechanism that determines which outcome you get isn't the processing code itself — it's whether you commit offsets at the right instant relative to the rebalance.

The Three Rebalance Callbacks and When They Fire

Confluent.Kafka exposes three hooks on ConsumerBuilder that let you react to changes in partition ownership: SetPartitionsAssignedHandler, SetPartitionsRevokedHandler, and SetPartitionsLostHandler. Each corresponds to a different transition in the consumer's relationship with its partitions, and each hands you different guarantees about what you can safely do inside it.

var consumer = new ConsumerBuilder<string, string>(consumerConfig)
    .SetPartitionsAssignedHandler((c, partitions) =>
    {
        // Fires when this consumer has just been handed a set of
        // partitions, before any message from them is delivered.
        // Consumption starts from the committed offset (or AutoOffsetReset).
        foreach (var tp in partitions)
            Console.WriteLine($"Assigned {tp.Topic} [{tp.Partition}]");
    })
    .SetPartitionsRevokedHandler((c, partitions) =>
    {
        // Fires BEFORE partitions move to another member, while this
        // consumer still legitimately owns them. This is your last safe
        // chance to persist progress on work already completed.
        try
        {
            // Commit() sends the offsets you STORED with StoreOffset after each
            // success. Do NOT commit 'partitions': their offsets are the
            // consumer's current position — one past the last message HANDED
            // to you, finished or not.
            c.Commit();
        }
        catch (KafkaException ex)
        {
            // e.g. Local_NoOffset when nothing new was stored since the last commit
            Console.WriteLine($"Commit during revoke failed: {ex.Error.Reason}");
        }
    })
    .SetPartitionsLostHandler((c, partitions) =>
    {
        // Fires when ownership was already taken away — a session timeout,
        // or max.poll.interval.ms exceeded. Another consumer may already
        // own these partitions, so a commit here would fail. Log it and
        // drop per-partition state; don't rely on it to save you.
        Console.WriteLine($"Lost {partitions.Count} partitions without a clean handoff");
    })
    .Build();

The distinction between revoked and lost is the one people miss. SetPartitionsRevokedHandler runs on the clean, negotiated path, under either assignment strategy (eager or cooperative, compared below) — the consumer is still an active, healthy group member and is voluntarily giving up partitions as part of a coordinated rebalance. SetPartitionsLostHandler runs when the consumer has already fallen out of the group involuntarily (for example, its session expired) and is finding out about it after the fact. Anything you attempt to commit in the lost handler is racing against a coordinator that may have already reassigned those partitions to someone else, so treat it as a place to log and clean up local state, not a place to rely on for correctness. Register both: if no lost handler is set, Confluent.Kafka calls the revoked handler for lost partitions too, so its commit runs against partitions you may no longer own. SetPartitionsAssignedHandler, by contrast, is where you find out what you're about to receive — useful for initializing per-partition state. If you're doing something unusual like resuming from a custom offset store, use the overload whose handler returns the starting offsets; calling Assign or IncrementalAssign (its cooperative-rebalancing counterpart) inside any rebalance handler throws an InvalidOperationException:

.SetPartitionsAssignedHandler((c, partitions) =>
    // Offset.Unset means "use the committed offset / AutoOffsetReset".
    partitions.Select(tp => new TopicPartitionOffset(tp, LoadCheckpoint(tp))))

All three handlers run on the thread that calls Consume() (or Close()), as a side effect of that call — never in the middle of your own message handler.

Why Commit Inside the Revoked Handler

Here's the concrete failure this handler exists to prevent. Say consumer A owns partition 3, and the last offset it committed was 101 — meaning everything up to and including offset 100 is durably accounted for. It then consumes and successfully processes messages 101 through 105 — writing to a database, sending a notification, whatever the handler does — but hasn't committed yet because its commit logic only runs every N messages or every few seconds.

Consumer A owns partition 3, committed offset = 101
 ↓
A processes offsets 101–105 (side effects already happened)
 ↓
Consumer B joins the group → rebalance triggered
 ↓
A's PartitionsRevokedHandler fires
 ↓
If A commits offset 106 here:
   Partition 3 → Consumer B
   B resumes from offset 106
   Messages 101–105 are NOT reprocessed
 ↓
If A does NOT commit here:
   Partition 3 → Consumer B
   B resumes from offset 101 (last commit before revoke)
   B reprocesses 101–105, duplicating every side effect A already did

Committing inside SetPartitionsRevokedHandler closes that gap. It's the last point in the partition's lifecycle on this consumer where a commit can still be expected to succeed — the consumer still owns the partition, so the broker accepts the commit, and it happens synchronously as part of the handoff rather than racing against whatever timer or batch size threshold your regular commit logic uses. Without it, every rebalance silently replays however much work happened since the last scheduled commit, on top of whatever gap your regular commit cadence already tolerates.

⚠️ Common Mistake: Treating the revoked handler as optional because "we commit periodically anyway." Periodic commits reduce the average reprocessing window; the revoked handler removes the rebalance-triggered one — in a single-threaded loop, down to zero finished-but-uncommitted messages, because the handler runs inside Consume(), between messages, when nothing is mid-processing.

It's worth being precise about what this does and doesn't guarantee. It covers the clean, negotiated handoff only. It does nothing for a crash, for an involuntary loss (session timeout or max.poll.interval.ms exceeded — the lost handler runs instead, and cannot commit), or for a commit that itself fails. And if you process messages on other threads, work can be in flight when the handler runs; then only offsets whose work has completed may be committed (see "Putting It Together" below). Duplicates therefore remain possible, which is why "At-Least-Once Is Not the Same as Safe" later in this lesson — and the "Delivery Semantics & Schema Strategy" lesson — treat them as something to handle downstream rather than something offset timing alone can eliminate.

Cooperative-Sticky vs Eager: How Much the Rebalance Disrupts

How disruptive a rebalance is — how many partitions get yanked away, and from how many consumers at once — depends on the PartitionAssignmentStrategy configured on the consumer. Confluent.Kafka's underlying librdkafka client defaults to an eager strategy (range or round-robin assignment) unless you explicitly opt into PartitionAssignmentStrategy.CooperativeSticky in ConsumerConfig. The difference between the two isn't just an implementation detail — it changes how much of your consumer group stalls every time membership changes.

Under an eager strategy, any rebalance is stop-the-world for the entire group: every member revokes all of its currently owned partitions, the group reforms, and a brand-new assignment is computed and handed out from scratch. Your PartitionsRevokedHandler receives the full set of partitions you owned, even the ones you'll get right back a moment later, and no partition in the group is being consumed during the reassignment window.

Under cooperative-sticky, the assignor tries to keep as much of the existing assignment intact as possible and only revokes the specific partitions that genuinely need to move to accommodate the change. If a new consumer joins and only two partitions need to shift to it, only the members giving up those partitions get a PartitionsRevokedHandler call, containing just those partitions; members that lose nothing get no revoke call at all, and keep consuming their untouched partitions without interruption. This matters concretely for a service with, say, twelve partitions and four consumers: with eager assignment, scaling to a fifth consumer freezes all twelve partitions across all five members for the duration of the rebalance; with cooperative-sticky, only the handful of partitions that actually change owners are paused.

For most production consumer groups — especially ones that scale up and down with load or redeploy frequently — cooperative-sticky is the better default, because it turns routine membership churn from a full-group stall into a localized, partial one. The trade-off is that the handlers become incremental: the revoked handler receives only the partitions being taken away (not "everything I had"), and the assigned handler receives only newly added partitions — possibly an empty list. Per-partition state must be added and removed incrementally rather than rebuilt from scratch. (Under the KIP-848 protocol, GroupProtocol = GroupProtocol.Consumer, rebalances are incremental by design and computed by the broker; PartitionAssignmentStrategy is not used.)

The Silent Rebalance: max.poll.interval.ms and Slow Processing

The rebalance scenario people find most confusing is the one where the consumer process is clearly alive — it hasn't crashed, its network connection is fine, and yet the broker reassigns its partitions anyway. This happens because Kafka doesn't consider "the process is running" sufficient evidence that your application is making progress; it considers "the application is calling Consume and returning control in a timely manner" the real signal of health.

In Confluent.Kafka, librdkafka runs group heartbeats on an internal background thread, decoupled from the thread that calls Consume. That's a deliberate design so a slow message handler doesn't immediately look like a dead consumer to the broker. But there's a second, independent guard: max.poll.interval.ms, exposed on ConsumerConfig.MaxPollIntervalMs, which defaults to five minutes. This setting tracks the time between successive calls back into the consumer's poll loop, and if your application spends longer than that processing a single message (or a batch) before calling Consume again, librdkafka proactively has the consumer leave the group — triggering exactly the same rebalance machinery as a crash, even though the heartbeat thread was reporting the process as alive the entire time.

Consumer calls Consume() → gets message at offset 105
 ↓
Handler starts a slow downstream call (e.g. a hung HTTP request)
 ↓
Heartbeat thread: still sending heartbeats, broker sees consumer as alive
 ↓
5 minutes pass, Consume() has not been called again
 ↓
max.poll.interval.ms exceeded → librdkafka leaves the group
 (no handler runs yet: handlers only run inside Consume(),
  and this thread is still stuck in the slow call)
 ↓
Partition reassigned to another consumer, which starts from the
last committed offset — at or before 105, so 105 is processed again
 ↓
The slow call finally returns. The next Consume() throws a
ConsumeException (ErrorCode.Local_MaxPollExceeded); the call after
that runs the PartitionsLostHandler (the assignment was lost, not
revoked); a commit now fails

This is the scenario the guidance about "process first, commit later" collides with directly: if a single message's processing time can exceed max.poll.interval.ms, you get an unwanted rebalance in the middle of legitimate work, and whichever consumer picks the partition up next will very likely redo that same message. Two practical levers exist. First, raise MaxPollIntervalMs to comfortably exceed your worst-case realistic processing time for whatever unit of work happens between Consume calls — but treat this as a ceiling for genuinely slow legitimate work, not a way to paper over a handler that's hanging indefinitely. Second, and usually the better fix, keep the work done per Consume call bounded — put a timeout on downstream calls, or offload genuinely slow operations (a large batch write, a call to a flaky downstream service) so the loop returns to Consume frequently, rather than blocking it for minutes at a time. Offloading has a catch covered in "Putting It Together": once work completes out of order, you must never store an offset past a message that hasn't finished.

💡 Mental Model: max.poll.interval.ms isn't asking "is the process alive?" — the heartbeat thread already answers that. It's asking "is the application still cycling back to ask for more work?" A consumer stuck on one slow message answers no to the second question even while answering yes to the first, and Kafka rebalances based on the second answer.

Put together, these three mechanics — the assigned/revoked/lost handlers, the choice of assignment strategy, and the poll-interval watchdog — form the actual timing surface that determines whether a rebalance in your service is a clean handoff or a source of duplicate side effects. Getting the commit call placed inside PartitionsRevokedHandler, choosing cooperative-sticky for groups that scale or redeploy often, and sizing MaxPollIntervalMs to your real processing time are the three concrete levers that convert "rebalances happen sometimes" from a source of production incidents into a well-understood, bounded event.

Putting It Together: The Complete Manual-Commit Loop

Here is the full call sequence this lesson has been building towards: store after every success, commit in batches, commit on revoke, and let Close() trigger the final commit.

var consumerConfig = new ConsumerConfig
{
    BootstrapServers = "localhost:9092",
    GroupId = "billing-service",
    AutoOffsetReset = AutoOffsetReset.Earliest,
    EnableAutoCommit = false,
    EnableAutoOffsetStore = false,
    PartitionAssignmentStrategy = PartitionAssignmentStrategy.CooperativeSticky
};

using var consumer = new ConsumerBuilder<string, string>(consumerConfig)
    .SetPartitionsRevokedHandler((c, _) =>
    {
        try { c.Commit(); }   // stored offsets = finished work only
        catch (KafkaException ex) { logger.LogWarning("Revoke commit: {Reason}", ex.Error.Reason); }
    })
    .SetPartitionsLostHandler((_, lost) =>
        logger.LogWarning("Lost {Count} partitions", lost.Count))
    .Build();

consumer.Subscribe("orders");

const int commitEvery = 100;
var sinceCommit = 0;
try
{
    while (!ct.IsCancellationRequested)
    {
        ConsumeResult<string, string> result;
        try
        {
            result = consumer.Consume(ct);
        }
        catch (ConsumeException ex) when (
            ex.Error.Code is ErrorCode.Local_KeyDeserialization or ErrorCode.Local_ValueDeserialization)
        {
            var bad = ex.ConsumerRecord;                 // raw bytes + coordinates
            await deadLetter.SendAsync(bad, ct);
            consumer.StoreOffset(new TopicPartitionOffset(bad.TopicPartition, bad.Offset + 1));
            continue;
        }
        catch (ConsumeException ex) when (!ex.Error.IsFatal)
        {
            // e.g. Local_MaxPollExceeded; a fatal error skips this catch and stops the loop
            logger.LogWarning("Consume error {Code} at {Position}",
                ex.Error.Code, ex.ConsumerRecord?.TopicPartitionOffset);
            continue;
        }

        await HandleOrderAsync(result, ct);   // throws → nothing stored for this message
        consumer.StoreOffset(result);         // stores result.Offset + 1

        if (++sinceCommit >= commitEvery)
        {
            try { consumer.Commit(); }        // sends every stored offset in one request
            catch (KafkaException ex) { logger.LogWarning("Batch commit: {Reason}", ex.Error.Reason); }
            sinceCommit = 0;                  // a failed batch is sent again with the next one
        }
    }
}
catch (OperationCanceledException) when (ct.IsCancellationRequested) { /* shutting down */ }
finally
{
    try { consumer.Close(); }   // runs the revoked handler (final Commit()), then leaves the group
    catch (KafkaException ex) when (ex.Error.IsFatal) { }   // a fatal client can't close; using disposes it
}

Four details carry the guarantee. StoreOffset runs only after HandleOrderAsync returns, so a stored offset always means "finished". Commit() sends stored offsets, never positions, so a batch commit can't overtake unfinished work. The revoked handler commits the same stored offsets before a partition moves, and Close() triggers it one last time on shutdown. A handler exception escapes the loop, finally commits everything that did finish, and the failed message is redelivered after restart; a production service would add bounded retries and a dead-letter route in front of that.

The alternative pairing is shorter: set EnableAutoCommit = true, keep EnableAutoOffsetStore = false, and delete the explicit Commit() calls. The timer commits stored offsets every auto.commit.interval.ms, and librdkafka also commits them when partitions are revoked and on Close(). The guarantee is the same; the replay window after a crash grows to one commit interval.

If you process in parallel, one rule matters: per partition, you may only store up to the first message that hasn't finished. Storing the offset of a later message that happened to finish first commits past the unfinished one — if the process then dies, that message is never redelivered. Either track completions per partition and store only the contiguous finished prefix, or Pause a partition while its work is running and Resume it afterwards, so you keep calling Consume() without receiving more of its messages.

✍️ Exercise: where does each failure resume?

The loop above has committed offset 5000 on partition 0, stored 5037 (messages 5000–5036 finished) and is handling 5037. For each event, which offset does the partition's next owner start from, and which messages are processed twice?

1. A cooperative rebalance moves partition 0 to another member.
2. The pod is killed with SIGKILL.
3. The handler for 5037 hangs for six minutes (MaxPollIntervalMs = 5 minutes), then returns.

Check your answer

1. The revoked handler runs inside the next Consume(), after 5037 has finished and 5038 has been stored. It commits 5038; the new owner starts at 5038 and nothing is repeated.
2. Nothing beyond 5000 was committed. The new owner starts at 5000 and processes 5000–5037 again (38 messages, including 5037 if it had completed).
3. After five minutes the client leaves the group and partition 0 moves; the new owner starts at 5000 and repeats 5000–5037. When 5037 finally returns, the loop stores 5038, but the next Consume() throws a non-fatal ConsumeException (Local_MaxPollExceeded), which the loop logs, and the call after that runs the lost handler — no commit is possible — so the stored 5038 is never sent.

Cases 2 and 3 are why the handler itself must be safe to run twice.

✍️ Harder variation: parallel completions

Partition 3 dispatches offsets 200–205 to worker tasks. So far 200, 201, 203, 204 and 205 have finished; 202 is still running. Which offset is safe to store now? What goes wrong if the code calls StoreOffset(resultFor205) as soon as 205 finishes, and the process then crashes?

Check your answer

Store 202 — the first unfinished message, which is also "the next offset to read". Everything below it is done; 202 itself is not.

Storing the result for 205 stores 206. The next Commit() (or revoke, or Close()) sends 206; if the process crashes before 202 completes, the group resumes at 206 and 202 is never processed — silent loss, produced by code that only ever stored offsets "after success". Once 202 finishes, the contiguous prefix jumps to 205 and you can store 206.

From Auto-Commit to At-Least-Once: Seeing the Guarantees Play Out

Configuration flags like EnableAutoCommit don't just toggle a setting — they decide which failure mode you're willing to live with. The cleanest way to see this is to run the same crash at two different moments in the same message's lifecycle and watch the outcome flip from silent loss to harmless-if-handled duplicate. The mechanics are simple enough to trace by hand, and once you've traced them once, you can recognize the same pattern in any consumer code you read afterward — auto-commit versus manual, timer versus outcome, silence versus echo.

Scenario 1: Auto-Commit Fires on a Clock, Not on Success

With EnableAutoCommit = true and EnableAutoOffsetStore left at its default true, the underlying librdkafka client commits the offsets of messages it has already handed to your code on a fixed timer — controlled by auto.commit.interval.ms, which defaults to five seconds. That commit happens whether or not your handler has finished, thrown an exception, or even started. The client doesn't know or care what your business logic did; it only tracks what offsets have been delivered to your Consume() calls.

Walk through a concrete timeline for a billing service consuming an orders topic:

T+0.0s  Consume() returns the message at offset 100 (a $500 charge request)
T+0.5s  Handler begins: call external payment gateway
T+5.0s  Auto-commit interval elapses -> client commits offset 101
        (offset 100 is now considered "done" from Kafka's point of view,
        even though the payment call hasn't returned yet)
T+6.2s  Payment gateway call is still in flight when the process crashes
        (out-of-memory kill, container eviction, unhandled exception in
        a background thread — the cause doesn't matter)

On restart, the consumer asks Kafka "where did I leave off?" and gets back offset 101, because that's what was last committed. The message at offset 100 is gone from this consumer's perspective — it will never be redelivered to this group. If the payment call had not actually completed before the crash, that $500 charge simply never happened, and nothing in the system will tell you. This is the at-most-once failure shape: the commit happened before (or independently of) completion, so a crash can leave work permanently undone with no signal.

It's worth tracing the same setup with the crash moved slightly earlier, because it shows why this bug is so easy to miss in testing:

T+0.0s  Consume() returns the message at offset 100
T+0.5s  Handler begins: call external payment gateway
T+4.2s  Process crashes — before the 5.0s auto-commit tick has fired

Here, offset 100 was never committed, so on restart it's redelivered and processed again — correct if the first gateway call never went through, a double charge if it had already succeeded just before the crash. Either way, nothing looks wrong in the logs. So auto-commit with auto-store gives neither guarantee: depending on where the crash lands relative to the timer, the same code loses a message or processes one twice. The only difference between this run and the earlier one is a two-second shift in when the crash happened relative to an invisible timer — and that timer is running independently of your code, your tests, and your intuition about "how long processing usually takes." A handler that's fast in staging and slow under production load (a saturated payment gateway, a cold connection pool, a GC pause) can silently cross from the safe case into the loss case without any code change at all. The interval doesn't need to be short for this to bite — it only needs to fire once between "message received" and "work actually finished."

Auto-commit has a second, less obvious trigger worth knowing about even though it's outside this lesson's focus on the manual-commit call sequence: a consumer group rebalance can also cause pending offsets to be committed as partitions are revoked, on whatever schedule the client considers "last known delivered." The exact mechanics differ from the interval timer, but the shape of the risk is identical — an offset can advance based on delivery, not completion. Whenever you see EnableAutoCommit = true, the question to ask is always the same one: "what moment does this commit correspond to, and is that moment guaranteed to be after my side effect finished?" If the answer is "no" or "I'm not sure," you're looking at loss risk, regardless of how the commit was actually triggered.

⚠️ Common Mistake: Treating auto-commit as "probably fine because it commits infrequently." Infrequent commits make the loss rarer, not impossible — rare enough to slip through testing, while the exposure stays in production under realistic latency.

Scenario 2: Commit Only After Success

Now crash the process again — this time just after the charge succeeds — but with EnableAutoCommit = false, EnableAutoOffsetStore = false and a handler that commits only once the payment call has returned successfully (the exact StoreOffset/Commit sequence is in "Putting It Together: The Complete Manual-Commit Loop" — here we're only tracing the outcome of committing after success).

T+0.0s  Consume() returns the message at offset 100 (a $500 charge request)
T+0.5s  Handler begins: call external payment gateway
T+2.1s  Payment gateway confirms the charge succeeded
T+2.3s  Process crashes before the commit call executes

Step through what happens on restart exactly the way the consumer would: it asks the broker for the last committed offset for this group and partition, gets back 100 (the commit that would have advanced it to 101 never happened), and resumes there. Kafka redelivers offset 100 — same key, same value, same partition, same offset number as the first delivery. The handler runs top to bottom again: deserialize the order, call the payment gateway, and this time the gateway call succeeds a second time, because from the gateway's point of view this looks like a brand-new charge request. Picture the ledger this produces:

Charge attempt #1 (first delivery,  T+2.1s): order-4471 charged $500, gateway ref A1B2
Charge attempt #2 (redelivery,      T+9.6s): order-4471 charged $500, gateway ref C3D4

Nothing was lost — the order was never silently skipped — but the customer now has two $500 charges tied to one order. This is the at-least-once shape: because you commit only after the work is done, a crash can never cause the offset to advance past unfinished work, so nothing silently disappears. The price is that the same message can be processed more than once.

It's also worth noting that "crash before commit" isn't the only path to this outcome. If the commit call is sent but the process dies (or the network stalls) before the broker's acknowledgment comes back, the client has no way to know whether the commit landed — so on restart it must, correctly, assume it didn't and reprocess from the last known-good committed offset. This doesn't change the guarantee at all; it's still at-least-once, and it's still handled the same way. It's mentioned here only to make the point concrete: there are more real-world timings that produce a redelivery than just "the crash happened one line before the commit statement." Anywhere completion and checkpointing are two separate steps, a gap exists, and that gap is what makes redelivery possible.

Two Crashes, Two Outcomes

The two crashes land at different points — mid-call in Scenario 1, just after the charge in Scenario 2 — so compare only whether the message comes back. That is decided by when the offset was allowed to move.

🎯 Key Principle: Delivery semantics aren't a property of Kafka the broker — they're a property of when your consumer chooses to move its checkpoint relative to when the work actually happened. Kafka will faithfully redeliver anything you haven't committed; whether that's a feature or a bug depends entirely on whether your commit was honest about what got done.

Recognizing the Regime You're In

Before moving on, it helps to fix a quick recognition habit, because in real codebases the configuration and the commit call are often far apart — a flag set in startup configuration, a commit buried inside a message loop dozens of lines later. Two questions settle it every time:

1. Who stores the offset? If EnableAutoOffsetStore is true (the default), offsets are stored the moment a message is handed to you — with EnableAutoCommit also true, commits follow a clock, not your handler, and you can both lose and replay messages whatever the handler looks like. If EnableAutoOffsetStore is false, only your StoreOffset calls make an offset committable — and the timer (if on) or your Commit() calls just send what you stored.
2. Where does the StoreOffset/Commit call sit relative to the side effect? If it's textually and logically after the point where the work is confirmed done (a returned success, a completed database write, a confirmed API response), you're in at-least-once territory. If a commit call appears before that confirmation — even "by accident," such as committing right after Consume() returns instead of after the handler runs — you've reintroduced at-most-once risk manually, without the auto-commit flag anywhere in sight.

That second trap is worth calling out explicitly: at-most-once loss isn't something only the auto-store-plus-auto-commit defaults can cause. Any code that advances the checkpoint before the work is verifiably finished has the same shape as Scenario 1, whether or not a timer is involved.

At-Least-Once Is Not the Same as "Safe"

It's tempting to treat "commit after success" as the finish line — you've avoided losing data, so surely you're done. But at-least-once only makes one promise: no message is skipped before its processing has finished — anything not yet done is delivered again. It makes zero promise about how many times that processing might run. Scenario 2 above didn't just illustrate redelivery in the abstract — it showed a real duplicate side effect: the payment gateway getting called twice for one order, with two distinct gateway references and two distinct charges landing on the same customer. If your handler's side effect isn't safe to repeat, at-least-once quietly converts "we never lose an order" into "we sometimes double-charge an order," which is not an obviously better trade for the customer on the receiving end.

The redelivered message is byte-for-byte identical to the first delivery — same key, same value, same offset. Kafka has no concept of "this one already happened once elsewhere"; that bookkeeping is entirely your responsibility. This is why idempotent processing isn't an optional polish step bolted on later — it's the other half of the at-least-once contract, and without it you haven't actually achieved a safe guarantee, just relocated the risk from "lost message" to "duplicate side effect."

It's also worth being precise about what "handling" a duplicate actually requires: it's not enough to detect that a message looks like one you've seen before, because two different orders can legitimately share every field except an ID, and two genuinely distinct deliveries of the same order need to collapse into a single effect even if they arrive far enough apart that no in-memory cache would still remember the first one. That's why the fix has to live in durable state — a database constraint or an idempotency key the downstream system honours, not a retry-loop guard or an in-process set of "recently seen" IDs that a restart would silently wipe.

For a side effect in a system you don't control, like a payment gateway, your database can't be updated atomically with the charge. The dependable tool is an idempotency key: most payment APIs accept one (Stripe's Idempotency-Key header, for example) and return the original result, instead of charging again, when they see the same key twice. Keys are kept only for a limited time (Stripe may prune them once they are 24 hours old), so if the process crashed between the charge and recording it, a redelivery that arrives after the key has expired — after a long outage, or a replay — can be charged again unless the handler first checks the gateway for an existing charge.

// Idempotent charge: the gateway deduplicates on a key derived from the order;
// our own table records the outcome once.
async Task HandleOrderCreatedAsync(ConsumeResult<string, string> result, CancellationToken ct)
{
    var order = JsonSerializer.Deserialize<OrderCreated>(result.Message.Value)!;

    // 1. Fast path: already completed on an earlier delivery?
    var done = await _db.ExecuteScalarAsync<bool>(
        "SELECT EXISTS (SELECT 1 FROM processed_orders WHERE order_id = @OrderId)",
        new { order.OrderId });
    if (done)
    {
        _logger.LogInformation(
            "Duplicate delivery for order {OrderId}, skipping charge", order.OrderId);
        return;
    }

    // 2. Charge with an idempotency key. If an earlier attempt reached the
    //    gateway before we crashed, the gateway returns that same charge.
    var charge = await _paymentGateway.ChargeAsync(
        order.OrderId, order.Amount, idempotencyKey: $"order-{order.OrderId}", ct);

    // 3. Record the outcome. A concurrent duplicate's insert becomes a no-op.
    await _db.ExecuteAsync(
        """
        INSERT INTO processed_orders (order_id, amount, gateway_ref, processed_at)
        VALUES (@OrderId, @Amount, @GatewayRef, now())
        ON CONFLICT (order_id) DO NOTHING
        """,
        new { order.OrderId, order.Amount, GatewayRef = charge.Reference });
}

The two mechanisms do different jobs. The idempotency key is what actually prevents the second charge: it covers a crash between step 2 and step 3, and the race where two instances handle the same order at once (one finishing a slow shutdown while another resumes from the same uncommitted offset) — the gateway still makes only one charge (Stripe, for example, rejects an overlapping request with a retryable conflict error; the retry gets the original charge). The unique constraint on order_id keeps your own record single. What would not work is the tempting reverse order — insert the "processed" row first, then charge: a crash between the two leaves the order marked done with no charge, and every redelivery then skips it. That is silent loss, the very thing at-least-once was supposed to prevent.

Not every side effect fits an "insert a new row" pattern, though. For a side effect like decrementing inventory, there is no natural row whose insertion is the side effect, but the same principle — let the database be the arbiter — still applies: insert a claim row, let it gate the update, and put both inside one transaction so they succeed or fail together:

// Idempotent inventory decrement: the adjustment row is the arbiter, and
// both statements run in one transaction on the same connection, so a
// crash can't split them. (_db is an open DbConnection; order is the event.)
await using var tx = await _db.BeginTransactionAsync(ct);

var claimed = await _db.ExecuteAsync(
    """
    INSERT INTO inventory_adjustments (order_id, sku)
    VALUES (@OrderId, @Sku)
    ON CONFLICT (order_id, sku) DO NOTHING
    """,
    new { order.OrderId, order.Sku }, tx);

if (claimed > 0)
{
    await _db.ExecuteAsync(
        "UPDATE inventory SET quantity = quantity - @Quantity WHERE sku = @Sku",
        new { order.Sku, order.Quantity }, tx);
}

await tx.CommitAsync(ct);

This one can be fully atomic, because the side effect and the dedup record live in the same database.

A dedup table like processed_orders, an idempotency key on an outbound API call, or a guarded update like the one above are the usual tools; which one fits depends on where the side effect actually lives — inside your own database, or on the far side of a third-party API you don't control.

Deciding Where Duplicates Are Tolerable

Not every handler needs this machinery, and reaching for a dedup table on every consumer is wasted effort in places where a duplicate genuinely doesn't matter. The decision cue is simple: ask what happens if this exact handler runs twice for the same message, back to back, with no changes in between.

The pattern underneath the table: side effects that are naturally idempotent by construction — setting a value, upserting a row to the same state, overwriting a cache key — can usually skip explicit dedup logic, because running them twice produces the same end state as running them once. Side effects that are inherently additive or one-shot — charging money, sending an email, decrementing a counter, calling a non-idempotent third-party API — are not duplicate-safe, because "twice" and "once" produce genuinely different outcomes. Whether they need a dedup key depends on what a duplicate costs: a double charge does, a second shipping notification usually doesn't.

💡 Mental Model: Ask "does running this twice change the state, or does it change the count of times something happened?" State-setting operations (upserts, overwrites) tend to be naturally duplicate-safe. Event-counting operations (charge, send, decrement) are not, and need a guard whenever a duplicate is costly.

Two quick worked examples show the cue in action. First: a handler that sends a "your order has shipped" push notification. Run it twice back to back and the customer's device shows the notification twice — mildly annoying, but the state of the world (the order is shipped, the customer knows it) is unchanged either way. That's a count-of-occurrences side effect, but a low-stakes one, so most teams tolerate it rather than build dedup infrastructure for it. Second: a handler that applies a one-time $10 signup credit to a customer's account by running UPDATE accounts SET balance = balance + 10 WHERE customer_id = @Id. Run that twice and the customer now has $20 of credit, $10 of which they were never supposed to receive — this looks like a state update syntactically, but semantically it's additive and one-shot, so it needs a durable guard: a credits_applied table keyed on (customer_id, promotion_id) with a unique constraint, inserted in the same transaction as the balance update, which runs only if that insert succeeds (the inventory pattern above). The lesson from comparing these two is that you can't tell which bucket a handler falls into from the statement's verb alone — UPDATE ... SET x = x + n is additive no matter how it's phrased, while UPDATE ... SET x = n is a genuine overwrite. Reading the side effect for whether repeating it changes an accumulated total is the reliable test, not skimming the verb.

✍️ Exercise: guard it or tolerate it?

For each handler, say whether running it twice for the same message is harmless, and if not, what guard you would add.

1. UPDATE customers SET email = @Email WHERE id = @Id
2. UPDATE loyalty SET points = points + 50 WHERE customer_id = @Id for a "first purchase" bonus event.
3. Calling an SMS provider to send "Your one-time login code is 481516".
4. Calling a shipping API that creates a pickup booking and accepts a client reference.

Check your answer

1. Harmless — an overwrite to the same value (as long as an older event can't arrive after a newer one; that is an ordering question, not a duplicate one).
2. Not harmless — additive. Record (customer_id, bonus_type) in a table with a unique constraint in the same transaction as the update, and apply the points only when the insert succeeds (the inventory pattern above).
3. Mostly tolerable — a second copy of the same code is annoying, not harmful; many teams accept it. If it must be exactly one, keep a sent-marker keyed on the event and accept that a crash between sending and marking can still produce a rare duplicate.
4. Not harmless — two pickups get booked. Pass a stable client reference derived from the order (the shipping API's idempotency mechanism), exactly like the payment example.

Seen side by side, the two crash scenarios reduce to one operating rule: commit timing determines whether a crash produces silence or an echo, and for most business services the echo — paired with an idempotent handler — is the far safer failure to design for.

Common Pitfalls and Pre-Production Checklist

Every one of the mistakes below is a real, recurring shape of failure in raw Confluent.Kafka consumers — not exotic edge cases, but the default outcome of copying a tutorial's config block and never revisiting it once the service is under load. None of them throw a compile error. All of them look fine in a demo. They surface weeks later as "we lost some orders" or "why did billing run twice."

Mistake 1: EnableAutoOffsetStore Left On While You Think You're Fully Manual

The section on consumer configuration explained what EnableAutoCommit and EnableAutoOffsetStore each disable internally. The pitfall is simpler than the mechanism: developers set EnableAutoCommit = false, see that auto-commit is gone, and assume they now have full manual control over offsets. They don't — because EnableAutoOffsetStore defaults to true in Confluent.Kafka's ConsumerConfig, mirroring the underlying librdkafka default of enable.auto.offset.store=true.

With that default left in place, librdkafka stores the offset of every message internally the moment Consume() returns it to your code — before your handler has done anything with it. EnableAutoCommit = false only stops that stored offset from being committed to the broker on a timer. If you later call consumer.Commit() anywhere in your code — during shutdown, in a periodic housekeeping task, or even accidentally in a retry path — it will commit whatever was auto-stored: at least the message currently in your handler, and, if you hand work to other threads, possibly many messages further. (Your own StoreOffset calls, meanwhile, throw a KafkaException, because librdkafka refuses manual stores while auto-store is on.)

// ❌ Wrong thinking: "I disabled auto-commit, so I have manual control."
var wrongConfig = new ConsumerConfig
{
    BootstrapServers = "localhost:9092",
    GroupId = "billing-service",
    EnableAutoCommit = false
    // EnableAutoOffsetStore left at its default (true) — offsets are
    // still being staged internally on every Consume() call, unprocessed.
};

// ✅ Correct thinking: both flags must be off for true manual control.
var consumerConfig = new ConsumerConfig
{
    BootstrapServers = "localhost:9092",
    GroupId = "billing-service",
    EnableAutoCommit = false,
    EnableAutoOffsetStore = false
    // Now nothing is stored or committed until your code explicitly
    // calls consumer.StoreOffset(...) after successful processing.
};

Where no StoreOffset call gives it away, the failure mode is quiet: the service looks like it's doing manual commits (because you wrote a Commit() call somewhere), but the offset being committed was already staged ahead of processing. Any Commit() that runs while a message is still being processed — a shutdown path, a periodic commit, a retry path — sends an offset past that message; if the process then dies before the message finishes, it is never redelivered. (A plain crash commits nothing when auto-commit is off; it is the explicit Commit() of an auto-stored offset that does the damage.) That is exactly the loss manual commits were supposed to eliminate.

💡 Pro Tip: If you're not sure whether both flags are set, log the resolved ConsumerConfig values at startup. Unless your code calls StoreOffset (which throws at once), a config typo here doesn't fail loudly; it fails quietly, months later, during an incident review.

Mistake 2: Catching ConsumeException Without Logging Where It Happened

The loop-building section covers how to distinguish transient from fatal errors via Error.IsFatal. The pitfall here is a habit that survives that correct logic: catching the exception, logging a generic message, and moving on — without capturing the coordinates of the failure.

// ❌ Wrong thinking: "I logged the exception, so I have visibility."
try
{
    var result = consumer.Consume(cancellationToken);
    ProcessMessage(result);
}
catch (ConsumeException ex)
{
    _logger.LogError(ex, "Error consuming message");
    // Which topic? Which partition? Which offset? Which key?
    // This tells an on-call engineer almost nothing at 2 AM.
}

A ConsumeException in Confluent.Kafka carries a ConsumerRecord on its .ConsumerRecord property in many failure cases (for example, deserialization failures), which exposes Topic, Partition, and Offset even though the message failed to deserialize. Losing that context means the only way to find the offending message later is to replay the whole partition and guess — effectively archaeology with a spoon, one offset at a time.

// ✅ Correct thinking: always log the coordinates that let you find
// the exact message again, even when the message body is unusable.
catch (ConsumeException ex)
{
    var record = ex.ConsumerRecord;
    _logger.LogError(ex,
        "Consume failed. Topic={Topic} Partition={Partition} Offset={Offset} " +
        "IsFatal={IsFatal} Reason={Reason}",
        record?.Topic,
        record?.Partition.Value,
        record?.Offset.Value,
        ex.Error.IsFatal,
        ex.Error.Reason);

    if (ex.Error.IsFatal)
    {
        throw; // stop the loop; this consumer instance can no longer make progress
    }
    // transient error: loop continues, next Consume() call will retry
}

The same discipline applies to a successfully deserialized message that fails in your own handler logic — log Message.Key, Partition, and Offset alongside the exception before routing to retry or DLQ. Without topic, partition, offset, and key on every failure log line, you cannot correlate a customer complaint ("my order didn't go through") with a specific Kafka record, and you cannot safely decide whether to replay, skip, or manually seek.

Mistake 3: Letting AutoOffsetReset Default to Latest on a New Group in Production

The configuration section covers what Earliest, Latest, and Error each mean for a group with no committed offsets. The production pitfall is not misunderstanding that behavior — it's never deciding it at all, and inheriting the librdkafka default of auto.offset.reset = latest.

Consider a concrete scenario: a team ships a new consumer group, inventory-sync-v2, to replace an older group. On first startup, the group has no committed offsets for any partition of the orders topic. With AutoOffsetReset left unset (defaulting to Latest), the new consumer starts reading only from the tail of the log — every order produced before the moment this consumer connected is skipped entirely, with no error, no warning, and no gap visible in the logs. The consumer looks healthy. Lag looks fine. The only symptom is a business one: inventory counts silently stop matching orders placed earlier that day.

// ❌ Wrong thinking: "I didn't set AutoOffsetReset, so Kafka will pick something safe."
var wrongConfig = new ConsumerConfig
{
    BootstrapServers = "localhost:9092",
    GroupId = "inventory-sync-v2",
    EnableAutoCommit = false,
    EnableAutoOffsetStore = false
    // AutoOffsetReset unset → defaults to Latest → backlog is skipped
    // on first run, silently, for a brand-new group.
};

// ✅ Correct thinking: choose the reset policy deliberately, based on
// whether this consumer needs history or only needs "from now on".
var consumerConfig = new ConsumerConfig
{
    BootstrapServers = "localhost:9092",
    GroupId = "inventory-sync-v2",
    EnableAutoCommit = false,
    EnableAutoOffsetStore = false,
    AutoOffsetReset = AutoOffsetReset.Earliest
};

This only matters for a new group, or one whose committed offsets are no longer usable (expired after the group sat empty, or pointing at data that retention has deleted) — once offsets are committed, AutoOffsetReset has no effect on normal restarts. But "new group" happens more often in practice than it sounds: renaming a GroupId during a refactor, standing up a parallel version of a service for a migration, or spinning up a consumer for a new downstream feature all create a brand-new group with no history, and each of those moments is exactly when a silent Latest default costs you a backlog you didn't know you were dropping. One caveat for a group that replaces an old one, as inventory-sync-v2 does: if the old group already applied these events, Earliest applies everything still retained a second time — so either, before its first start, set the new group's offsets to where the old one stopped (kafka-consumer-groups --reset-offsets can do this), or make sure the handler is idempotent first.

Mistake 4: Long Synchronous Work Inside the Poll Loop

Rebalance handlers and max.poll.interval.ms are covered in depth in the section on partition assignment. The pitfall to recognize here is a specific, common shape of code: doing slow, blocking work directly inside the loop between calls to Consume().

// ❌ Wrong thinking: "heartbeats are handled in the background, so
// slow processing in my loop is safe."
while (!cancellationToken.IsCancellationRequested)
{
    var result = consumer.Consume(cancellationToken);

    // Blocking, synchronous call to a slow downstream system —
    // no timeout, no async, potentially minutes under load.
    var invoiceResponse = billingClient.ChargeCustomerBlocking(result.Message.Value);

    consumer.StoreOffset(result);
    consumer.Commit();
}

In Confluent.Kafka, heartbeats to the group coordinator are indeed sent from a background thread, so the connection to the broker doesn't visibly drop while this loop is busy. But the client itself still requires the application to call Consume() again within max.poll.interval.ms (default measured in minutes, but very achievable to exceed under load or a downstream outage). If your synchronous work inside the loop runs longer than that window, librdkafka proactively sends a "leave group" on your behalf — even though heartbeats kept flowing — and the coordinator reassigns your partitions to another member.

Here is what that produces end to end:

Consume() returns message at offset 104
  ↓
Handler starts a slow, blocking charge call
  ↓
Time exceeds max.poll.interval.ms
  ↓
librdkafka leaves the group on this member's behalf
  ↓
Partition reassigned to another consumer instance
  ↓
Original handler eventually finishes charging the customer
  ↓
Original consumer tries to commit offset 105 — but it is no longer a
group member, so the commit throws a KafkaException
  ↓
New owner starts from the last successfully committed offset (104),
re-reads offset 104, and charges the customer again

The result is a duplicate side effect — a double charge — that has nothing to do with a crash or a network blip. The consumer process never went down; it just took too long between polls, so the client gave up its place in the group and the partition moved. This is why the recommended shape keeps the loop itself light: put timeouts on downstream calls, or hand the message to bounded async work and return to Consume() promptly — storing offsets only up to the first unfinished message, as described in "Putting It Together" — and tune max.poll.interval.ms upward only as a deliberate accommodation for known-slow work rather than as a blanket fix for unbounded processing time.

Pre-Production Checklist

Those four mistakes span three parts of the consumer's lifecycle — configuration, loop error handling, and rebalance timing — which is exactly why they're easy to miss individually even when each piece looks correct in isolation. Before relying on manual commits and graceful shutdown in production, walk through all three areas together.

✍️ Exercise: review this consumer

A pull request contains this consumer. Find at least four problems before opening the answer.

var config = new ConsumerConfig
{
    BootstrapServers = "kafka:9092",
    GroupId = $"invoices-{Guid.NewGuid()}",
    EnableAutoCommit = false
};

using var consumer = new ConsumerBuilder<string, string>(config)
    .SetPartitionsRevokedHandler((c, partitions) => c.Commit(partitions))
    .Build();
consumer.Subscribe("invoices");

while (true)
{
    try
    {
        var r = consumer.Consume(ct);
        await SendInvoiceEmailAsync(r.Message.Value);
        consumer.Commit(new[] { r.TopicPartitionOffset });
    }
    catch (Exception ex)
    {
        logger.LogError(ex, "Consume failed");
    }
}

Check your answer

• GroupId with a GUID: every pod — and every restart — is a brand-new group that reads all partitions, and…
• no AutoOffsetReset: …each of those new groups starts at Latest, skipping everything already in the topic.
• EnableAutoOffsetStore left on: harmless in this exact code only because nothing calls StoreOffset or a bare Commit() — the first person to add a bare Commit() commits offsets stored on hand-off, finished or not (and a StoreOffset call would throw).
• Revoked handler commits partitions: those are the consumer's positions, not finished work — it can commit past a message whose handler failed. And with no lost handler registered, Confluent.Kafka calls this handler for lost partitions too, committing to partitions another member may already own.
• Off-by-one: Commit(new[] { r.TopicPartitionOffset }) commits the record's own offset, so with a fixed GroupId the last record would be replayed after every restart. Use Commit(r).
• Blanket catch (Exception) inside while (true): on shutdown, the cancelled token makes Consume throw OperationCanceledException on every pass — a hot loop that never exits; it also logs no topic, partition or offset, and treats fatal errors like transient ones.
• No Close() (and no way out of while (true) to reach one): the final commit never runs, and with a shared GroupId the group would wait out the session timeout on every deploy.
• A failed email is skipped silently: the exception is caught and the loop moves on to the next message, whose later commit advances past the failed one.

⚠️ Common Mistake: treating this as a one-time setup checklist rather than a code-review checklist. Config drifts (someone "simplifies" the ConsumerConfig during a refactor), new handlers get added to the loop without anyone checking their worst-case latency, and GroupId gets changed during a migration without anyone re-deciding AutoOffsetReset. Re-run this checklist whenever any of those three areas changes, not just before the first deployment.

💡 Real-World Example: a service that processes payment webhooks moved its handler from a fast in-memory validation to a synchronous call against a fraud-scoring API. Nothing else in the code changed. Under normal load the call returned in milliseconds; under a fraud-provider slowdown it occasionally took long enough to exceed the poll interval, triggering rebalances and duplicate charge attempts that only idempotent downstream logic prevented from reaching the customer twice. The lesson isn't "never call external services from a handler" — it's that any change to per-message processing time needs to be checked against max.poll.interval.ms, not assumed safe because it was fast yesterday.

With configuration, loop error handling, and rebalance timing checked together, you've closed the gaps that don't show up in a demo but reliably show up in production: offsets committed ahead of processing, failures nobody can trace back to a specific record, backlogs silently skipped by a new group, and duplicate side effects from rebalances that had nothing to do with a crash. As a next step, run the crash-before-commit test and the two-consumer rebalance test against your actual ConsumerConfig and handler code — not a simplified test double — since several of these pitfalls (the offset-store flag in particular) only manifest under those exact conditions.