Core Kafka Mental Model

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

• predict what happens to a record after it has been read, and what actually removes it;
• say exactly what a completed ProduceAsync promises for a given acks / min.insync.replicas combination;
• trace a consumer loop through a crash and say whether a record is lost, replayed, or handled twice;
• decide when ordering holds and when it does not;
• recognise the four recurring mental-model mistakes from their symptoms in logs and metrics.

You need C# and async/await; no prior Kafka knowledge is assumed. Each section includes a short exercise — try it before opening the answer.

📹Watch the Lesson Video20 min

The Log Is the Data Structure

Every developer who has run SELECT * FROM Orders WHERE Id = 42 and gotten one row back carries an instinct: data lives somewhere, identified by a key, and you go fetch it. Every developer who has used a work queue carries a second instinct: a message is a unit of work that one worker takes, processes, and destroys. Kafka breaks both instincts, and almost every confused Kafka design traces back to a developer still holding one of them.

Kafka's storage primitive is neither a table nor a queue. It is an append-only log: an ordered, immutable sequence of records (individual messages) that you read by position, not by predicate and not by popping. Once that single structure feels boring — genuinely obvious, the way a List<T> is obvious — the rest of Kafka stops being a collection of trivia and starts being consequences. This section builds the log from first principles: how records get in, what a record actually is to a broker (a Kafka server), how reading works, and how records eventually leave.

Appending to the tail

Picture a text file you can only do two things with: append a line to the end, or read line N. You cannot edit line 5. You cannot delete line 5. You cannot insert between lines 5 and 6. That constraint is the design.

When a producer (the application code that writes records) sends a record, the broker appends it to the end of a log file and assigns it the next integer in sequence. That integer is the offset. Offsets start at 0 and increase monotonically by one per record within a single log.

offset:   0      1      2      3      4      5   ← next append lands here
        ┌──────┬──────┬──────┬──────┬──────┐
        │ rec  │ rec  │ rec  │ rec  │ rec  │
        └──────┴──────┴──────┴──────┴──────┘
         oldest                      newest (the "tail")

The offset is not a database identity column and not a globally unique message ID. It is a coordinate: the address of a slot in one specific log. Offset 4 in one log has nothing whatsoever to do with offset 4 in another. This matters the moment you have more than one log — which in Kafka you always do, since a topic — a named stream of records, such as orders — is split into partitions, each partition being an independent log. The partitioning story belongs to later lessons; what you need here is that "the log" is the unit that offsets and ordering are defined against.

Immutability buys three things at once. Writes are pure appends, so the broker never seeks to rewrite an existing byte range — the performance consequences of that are taken up in "Brokers, Replication, and Durability Guarantees." Replication — keeping copies of each log on other brokers — becomes trivially defined: a follower (a broker holding a copy) is caught up if it holds the same bytes up to the same offset. And readers get a stable world — a record you read at offset 100 will still be byte-identical at offset 100 tomorrow, which is what makes replaying history meaningful.

🎯 Key Principle: An offset is a position in one log, permanently bound to one record. It is never reused, never reordered, and never reassigned (short of the unclean leader election noted at the end of this section).

What a record actually is

A Kafka record has four parts:

Notice what is not in that list: a schema, a type name, a routing rule, a priority, a TTL, a delivery counter. The broker treats key and value as opaque byte arrays. It does not parse your JSON. It does not know you are using Avro or Protobuf. It will happily accept a value that is a corrupted half-serialized object and hand it to every consumer (an application that reads the log), forever.

This is a genuine design decision, not an oversight. A broker that never parses payloads does no work that depends on what they mean — it checks record-batch checksums and batch metadata (producers send records in batches), but never deserializes a value — which is what allows it to move data from disk to socket with minimal handling. The cost is that payload correctness is entirely your application's problem — serialization lives in the .NET client, which is why the write and read paths in "Producers, Consumers, and the Shape of a Kafka Application" both start with a serializer choice.

The one exception worth knowing: the key is not entirely opaque in behavior, even though it is opaque in content. The broker compares key bytes for equality during log compaction, and the client hashes key bytes to choose a partition. Neither operation requires understanding what the bytes mean.

💡 Real-World Example: A team ships an order-events topic where the value is JSON. Someone deploys a service that accidentally writes the string "null" instead of a JSON object. Kafka accepts every one of those records without complaint — they are valid bytes. The failure surfaces days later in a downstream consumer's deserializer, and because the records are immutable, the bad ones cannot be edited out. They stay until retention removes them, and every consumer that replays that range must be able to skip them.

Reading is a cursor, not a pop

Here is the conceptual pivot. In a traditional queue, Receive() removes. The broker mutates its own state: this message is now in-flight, now acknowledged, now gone. State about your progress lives inside the broker, per message.

In Kafka, reading is a fetch by offset. A consumer says "give me records starting at offset 250 in this log," and the broker returns bytes. The log is unchanged. Nothing was consumed in the destructive sense of the word. The consumer's progress is nothing more than an integer it keeps track of — a cursor. (It saves that integer to the cluster from time to time so it can resume after a restart; the broker stores the number but never uses it to decide what to keep.)

The direct consequence is that independent readers cost the log nothing:

Log:  [0][1][2][3][4][5][6][7][8][9][10][11][12]

Analytics service   → cursor at offset 3   (replaying history, way behind)
Fraud detection     → cursor at offset 11  (near real time)
Audit exporter      → cursor at offset 7   (batch, hourly)
New service (today) → cursor at offset 0   (rebuilding state from scratch)

All four read the same bytes. None affects any other. The fraud service crashing does not hold back the analytics service. Adding a fifth reader tomorrow requires zero change to the producer and zero change to existing consumers — it just starts reading at whatever offset it chooses.

This is why Kafka is described as a publish-subscribe log rather than a queue. Fan-out is free because reads are non-destructive; adding a consumer adds read load, not bookkeeping.

Here is the shape of that idea as plain .NET, deliberately without any Kafka client — just the data structure:

// A toy in-memory model of the log primitive. No Kafka client involved —
// this is the data structure Kafka's behavior follows from.
public sealed record LogRecord(byte[]? Key, byte[]? Value, long TimestampMs);

public sealed class AppendOnlyLog
{
    private readonly List<LogRecord> _records = new();

    // Producers only ever add to the tail. The returned offset is the
    // position the record was written to — it never changes afterwards.
    public long Append(LogRecord record)
    {
        _records.Add(record);
        return _records.Count - 1;
    }

    // Reading takes a position and returns a window. Nothing is removed,
    // and the log has no idea who called this or how often.
    public IReadOnlyList<LogRecord> Read(long fromOffset, int maxRecords)
    {
        if (fromOffset >= _records.Count) return Array.Empty<LogRecord>();
        var count = (int)Math.Min(maxRecords, _records.Count - fromOffset);
        return _records.GetRange((int)fromOffset, count);
    }
}

The important detail is what AppendOnlyLog does not have: no Dequeue, no Acknowledge, no per-consumer state, no Delete(offset). A reader is just someone holding an integer:

// Two independent readers over the same log, at different positions.
var log = new AppendOnlyLog();
for (int i = 0; i < 100; i++)
    log.Append(new LogRecord(null, BitConverter.GetBytes(i), 0));

long analyticsCursor = 0;    // replaying from the beginning
long realtimeCursor  = 95;   // only cares about recent records

var batch = log.Read(analyticsCursor, 10);
analyticsCursor += batch.Count;      // now 10 — the reader advanced itself

var tail = log.Read(realtimeCursor, 10);
realtimeCursor += tail.Count;        // now 100 — caught up to the tail

// The log still holds all 100 records. Reading removed nothing.
// Rewinding is trivial: analyticsCursor = 0 replays everything again.

That last line is the superpower. Reprocessing a month of events after fixing a bug is cursor = someEarlierOffset, not a data-recovery project. (Simplified, of course: real consumers store their cursor durably and coordinate it across instances — that's the subject of the "Consumer Groups & Offsets" lesson.)

🧠 Mnemonic: Queues pop, logs point. A queue hands you a message and forgets it. A log hands you a position and remembers everything.

Three structures, side by side

The fastest way to solidify this is to compare against the two things learners keep mistaking Kafka for.

The "progress state" row is the one with the biggest operational consequence. A queue broker must remember, for every single message, whether it was delivered, whether it was acknowledged, how many times it was retried, and when its visibility timeout expires. That is per-message mutable state, and it is why queue brokers do real work proportional to message count. A Kafka broker remembers one number per consumer group — the instances of one service that share a topic's work — per log (the committed offset: where that group resumes). Ten million unread records cost the broker no more bookkeeping than ten.

❌ Wrong thinking: "My consumer processed the record, so it's gone from Kafka now." ✅ Correct thinking: "My consumer moved its cursor past the record. The record is still on disk and another consumer — or my consumer after a rewind — can read it again."

🤔 Did you know? Because the log is the source of truth and readers are just cursors, a Kafka topic can serve as the input to rebuilding a database rather than a copy of one. Replaying a compacted topic (one the broker trims to the latest record per key, covered below) from the beginning reconstructs the latest value per key — the log becomes the authoritative state and the database becomes a derived, disposable view. This inversion, sometimes called "turning the database inside out," is a direct consequence of non-destructive reads, not a bolt-on feature.

The only way out: retention

If records are never removed by reading, disks would fill forever. They don't, because deletion is governed by an entirely separate mechanism: retention, configured per topic and enforced by the broker on its own schedule, with no knowledge of who has or hasn't read anything.

There are three removal mechanisms, configured through two cleanup policies (delete, which covers both the time and size limits, and compact):

🕒 Time-based deletion (retention.ms) — segments older than the configured age are deleted. Retention operates on log segments (the files a log is chopped into), so deletion happens in chunks at segment granularity, not record by record. This is why a record slightly older than the retention window can still be readable for a while.

📏 Size-based deletion (retention.bytes) — once the log exceeds a byte ceiling, the oldest segments are removed. When both time and size limits are configured, whichever triggers first wins.

🔑 Compaction (cleanup.policy=compact) — instead of dropping the oldest records, the broker keeps at least the latest record for each key and eventually removes superseded earlier records with the same key. A record with a non-null key and a null value is a tombstone, marking the key for eventual removal. Compaction turns a log into something closer to a changelog of a key-value store: replay it from the start and you get current state per key. The cleanup.policy setting accepts delete, compact, or both combined.

Before compaction (keys shown above values):
  offset:   0      1      2      3      4      5
  key:      A      B      A      C      A      D
  value:    v1     v1     v2     v1     v3     v1

After compaction — latest value per key survives:
  offset:   1      3      4      5     ← 0 and 2 are gone; survivors keep their offsets
  key:      B      C      A      D
  value:    v1     v1     v3     v1

Note what compaction does not preserve: dense, contiguous offsets. Surviving records keep their original offsets, so a compacted log has gaps. A consumer that assumes "the next record is always at offset+1" breaks here; consumers must treat offsets as increasing, not consecutive. (Gaps appear without compaction too: the commit and abort markers of transactions, covered later, occupy offsets that consumers never see.)

⚠️ Common Mistake: Assuming a slow consumer is safe because "the messages are waiting for it." They aren't waiting — retention runs on wall-clock time and byte counts, indifferent to consumer lag (how far a consumer's cursor trails the newest record). A consumer down for longer than the retention window comes back to find its cursor pointing at an offset that no longer exists, and depending on the client's auto.offset.reset behavior it will jump to the earliest or latest available offset — silently skipping records in the latest case. In Confluent.Kafka (the .NET Kafka client, which wraps the C library librdkafka) the default for auto.offset.reset is latest, so the silent-skip outcome is the one you get unless you set it explicitly.

💡 Mental Model: Think of the log as a security-camera recording with a fixed-length tape, not an inbox. Watching the footage doesn't erase it; the tape overwrites on its own schedule regardless of whether anyone watched. Your job as a consumer is to keep up with the tape, not to empty the inbox.

✍️ Exercise: trace a compaction

A compacted topic's partition holds these records, all in closed segments (the broker never compacts the active segment — the file still being appended to — so assume it has already rolled over to a new one). ∅ is a null value.

offset:  0    1    2    3    4    5    6
key:     A    B    C    A    B    A    D
value:   v1   v1   v1   v2   ∅    v3   v1

1. Which offsets survive a compaction run, and what is special about one of the survivors?
2. A brand-new consumer replays the partition from the start afterwards. What state does it rebuild for keys A–D?
3. A consumer that read offsets 0–2 before compaction resumes at offset 3 afterwards. Does it miss anything?

Check your answer

1. Offsets 2, 4, 5, 6 survive: the latest record per key (C@2, B@4, A@5, D@6). Offsets 0, 1 and 3 are superseded and removed, and the survivors keep their original offsets — so the log now has a hole at 3, between two survivors. Offset 4 is a tombstone: it is kept for a while (the topic's delete.retention.ms) so that consumers get a chance to see the deletion, and is removed by a later compaction run.
2. A = v3, B deleted, C = v1, D = v1. If the tombstone has already been cleaned up by then, the replaying consumer never sees B at all — which gives the same final state.
3. Offset 3 (A = v2) no longer exists, so its fetch from offset 3 simply returns the next record that does: offset 4. It never sees A = v2 — but that value was superseded by A = v3 at offset 5, which it does see, along with the tombstone for B and D = v1. It ends with the same state as everyone else. Seeing that tombstone is the reason tombstones are kept for a period instead of vanishing immediately.

(Two operator-level exceptions exist: an administrator can explicitly delete records up to an offset with the DeleteRecords admin API, and an unclean leader election — letting a copy of the log that had fallen behind take over after a broker failure (a replica outside the ISR; the next section explains both) — can truncate records. Neither is triggered by reading.)

Everything else in Kafka — replication, consumer groups, the delivery guarantees you choose, transactions — is machinery built to preserve or exploit these properties: appends at the tail, opaque bytes, monotonic offsets, cursor-based reads, and retention-driven removal.

Brokers, Replication, and Durability Guarantees

The log from the previous section lives somewhere physical: on files, on machines, in a datacenter that occasionally loses a rack. The question this section answers is deceptively simple — when your .NET producer's ProduceAsync task completes without throwing, what exactly has been promised to you? The honest answer ranges from "nothing at all" to "this record survives the loss of any single machine," and the difference is a handful of configuration keys you choose deliberately.

The Broker and the Cluster

A broker is one Kafka server process. Its job is unglamorous: own a set of partition replicas (its copies of partition logs), write incoming records to log segment files on local disk, serve fetch requests from those files, and keep its copy of the cluster metadata up to date. If you SSH into a broker and look at its log directory, you see exactly the structure you'd expect from the append-only model:

/var/lib/kafka/data/
  orders-0/                       ← topic "orders", partition 0
    00000000000000000000.log      ← segment: records starting at offset 0
    00000000000000000000.index    ← offset → byte-position lookup
    00000000000000000000.timeindex
    00000000000000524288.log      ← next segment, rolled when the prior filled
  orders-3/
  payments-1/

The partition is the unit of everything. A topic named orders with 6 partitions is six independent logs, and each of those six is placed on some broker as a leader with copies on other brokers as followers. A cluster is just the set of brokers that share a common metadata layer — they agree on which topics exist, how many partitions each has, and which broker currently leads each partition. That metadata is owned by the controller: in Kafka 4.x (KRaft mode — Kafka's built-in metadata quorum — the only mode since ZooKeeper, the separate coordination service older versions needed, was removed) a small quorum of controller nodes keeps it as a replicated log of its own, and the active controller reacts to broker failures and assigns new partition leaders. Brokers follow that metadata log; they do not gossip it among themselves. The "KRaft & Modern Kafka" lesson covers the quorum in detail.

🎯 Key Principle: Brokers do not know about each other's data contents. They know about assignments. Broker 2 knows it leads orders-0 and that brokers 3 and 5 replicate it — it does not know or care what your order payloads contain.

Leader and Follower Replication

Each partition has a replication factor — the number of copies the cluster maintains. With replication.factor=3, partition orders-0 exists three times on three different brokers. Exactly one of those copies is the leader at any moment.

Partition orders-0, replication.factor = 3

Producer
   ↓  (writes always go to the leader)
Broker 2 : orders-0 LEADER    log = [0,1,2,3,4,5,6,7]
   ↓  (followers FETCH from the leader — a pull, not a push)
Broker 3 : orders-0 FOLLOWER  log = [0,1,2,3,4,5]   fetching, in ISR
Broker 5 : orders-0 FOLLOWER  log = [0,1,2,3]       stalled, dropped from ISR

ISR = in-sync replicas: the copies keeping up with the leader (defined below)

The replication mechanism is worth internalizing because it is the same mechanism consumers use. A follower is essentially a consumer of the leader's log: it issues fetch requests saying "give me everything from offset 6 onward," appends what it receives to its own copy of the segment files, and repeats. There is no separate replication protocol, no two-phase commit, no consensus round per record. The log's append-only, offset-addressed shape makes replication a byte-range copy.

Producers and consumers both talk to the leader by default. A follower that has fallen behind is not serving anyone; it is quietly catching up. If broker 2 dies, the controller elects a new leader for orders-0 from the replicas that were keeping up with it (the in-sync replicas, defined next), and clients discover the new leader through a metadata refresh — which is why the .NET client's ProduceAsync may briefly surface a NotLeaderForPartition condition internally and retry rather than failing your call.

In-Sync Replicas and the High Watermark

A follower counts as in-sync while it keeps up with the leader — the test is time-based, controlled by replica.lag.time.max.ms (30 seconds by default), not "does it hold every byte the leader holds right now." A follower a few records behind that keeps fetching and catches up to the leader's end of log within that window is in-sync; one that stops fetching, or keeps fetching but has not caught up for longer than the window, is removed. That set is the in-sync replica set, or ISR, and it always includes the leader. In the diagram above the ISR is {2, 3} — broker 5 is a replica but has stalled out of it.

From the ISR, the leader computes the high watermark: the first offset that is not yet on every member of the ISR. Everything below it is on all of them. This single number is the most important boundary in Kafka's durability story.

Leader log:  [0][1][2][3][4][5][6][7]
                               ↑
                      high watermark = 6
              (offsets 0–5 are on every in-sync replica — broker 3
               has them too; 6 and 7 exist only on the leader so far)

A consumer fetching this partition can read offsets 0 through 5.
Offsets 6 and 7 are invisible to it — not yet on every in-sync replica.

⚠️ Common Mistake: Assuming a record is readable the instant the leader writes it. It is not. Kafka deliberately withholds records above the high watermark from consumers, because those records could still be lost if the leader fails before replication completes. This is why any record a consumer successfully fetches is already on every in-sync replica — a guarantee you get for free and should not try to re-implement. Mind how many that is: if the ISR has shrunk to the leader alone, it is one copy — the case the min.insync.replicas floor below exists for. ("On" means held by each ISR member, typically in its page cache — the operating system's in-memory copy of recently written file data; Kafka relies on replication, not a per-record disk flush, for durability.)

💡 Mental Model: The high watermark is a second cursor on the log, owned by the cluster rather than by any client. Producers push the tail forward; replication pulls the high watermark along behind it; consumers are only allowed to walk up to the high watermark.

The acks Spectrum: What "Written" Means

The producer decides how much of the replication process it waits for. In Confluent.Kafka this is the Acks property on ProducerConfig, which maps to librdkafka's acks setting. librdkafka already defaults to acks=all; setting it explicitly documents the intent and protects you from a shared config that overrides it.

using Confluent.Kafka;

// Fire-and-forget: the lowest-latency, weakest-guarantee setting.
var fireAndForget = new ProducerConfig
{
    BootstrapServers = "broker1:9092,broker2:9092", // brokers to contact first; the client discovers the rest
    Acks = Acks.None            // acks=0 — no broker response at all
};

// Leader-only: broker confirms its own local write, not replication.
var leaderOnly = new ProducerConfig
{
    BootstrapServers = "broker1:9092,broker2:9092",
    Acks = Acks.Leader          // acks=1
};

// Full ISR: broker responds only after all in-sync replicas have the record.
var durable = new ProducerConfig
{
    BootstrapServers = "broker1:9092,broker2:9092",
    Acks = Acks.All,            // acks=all
    EnableIdempotence = true,   // safe retries; detailed in a later section
    MessageSendMaxRetries = int.MaxValue,
    MessageTimeoutMs = 120_000  // upper bound on how long a send may be retried
};

What each level actually buys you:

⚠️ Common Mistake: Believing acks=all alone guarantees redundancy. It does not. acks=all means "all replicas currently in the ISR." If both followers of a three-replica partition have fallen behind and dropped out, the ISR is {leader}, and acks=all degrades silently to acks=1. You get the latency of a strong setting and the durability of a weak one.

min.insync.replicas: The Broker-Side Floor

The fix is a broker/topic-level setting, not a producer one. min.insync.replicas declares the minimum ISR size at which the partition will still accept acks=all writes. If the ISR shrinks below that number, the leader rejects the write with NOT_ENOUGH_REPLICAS rather than accepting something it cannot make durable.

The canonical production configuration is replication.factor=3 with min.insync.replicas=2:

replication.factor = 3, min.insync.replicas = 2, producer acks = all

ISR = {L, F1, F2}  → write accepted, needs 3 acks   ✅ stays writable after 1 broker loss
ISR = {L, F1}      → write accepted, needs 2 acks   ✅ still redundant
ISR = {L}          → write REJECTED (NotEnoughReplicas) ❌ availability sacrificed

That last line is the trade-off stated bluntly: the partition becomes unwritable rather than accepting a record that lives on exactly one machine. Setting min.insync.replicas=3 with replication.factor=3 looks stronger but is usually worse — a single broker restart for a routine patch takes the partition offline for writes.

💡 Pro Tip: In your .NET producer, treat ErrorCode.NotEnoughReplicas and ErrorCode.NotEnoughReplicasAfterAppend (the leader appended the record, but the ISR shrank before enough copies existed) as retriable-but-alertable. The client will retry them, but their appearance means your cluster is running with reduced redundancy right now.

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

try
{
    var result = await producer.ProduceAsync(
        "orders",
        new Message<string, string> { Key = orderId, Value = payloadJson });

    // result.Status == PersistenceStatus.Persisted means the broker confirmed
    // the write at the level demanded by Acks. With Acks.All that implies
    // the record is on every in-sync replica.
    logger.LogInformation(
        "Wrote to {Topic}/{Partition} at offset {Offset}, status {Status}",
        result.Topic, result.Partition.Value, result.Offset.Value, result.Status);
}
catch (ProduceException<string, string> ex)
{
    // PossiblyPersisted: the broker may have written it but we never saw the ack.
    // Idempotence makes the client's OWN retries safe; by the time this reaches you,
    // those retries are exhausted, and resending is a new record Kafka won't dedupe.
    logger.LogError(ex, "Produce failed: {Reason} (status {Status})",
        ex.Error.Reason, ex.DeliveryResult.Status);
}

The PersistenceStatus enum on the delivery result is worth knowing by name: Persisted, NotPersisted, and PossiblyPersisted. That third value is not a bug — it is the network being honest about the ambiguity of a lost acknowledgement. Idempotent producer sequence numbers (per-partition counters that let the broker spot a resent batch) remove that ambiguity for the retries the client makes on its own (see "Delivery Semantics and Ordering: What Kafka Actually Promises"); once PossiblyPersisted reaches your code, deciding whether to resend — and making a possible duplicate harmless downstream — is your application's job.

Why Disks Are Not the Bottleneck

A reasonable objection: copying every record to three machines before acknowledging sounds slow. In practice it isn't, and the reason is that Kafka's access pattern is the one storage hardware is best at.

🔧 Sequential append. Kafka never seeks to update a record in place. Every write goes to the end of the active segment file. On spinning disks sequential throughput is orders of magnitude higher than random access, because there is no head movement; on SSDs the gap is much smaller, but large sequential writes still avoid read-modify-write amplification. A B-tree-backed queue that mutates rows to mark them consumed pays random-I/O cost on every message; Kafka pays none.

🔧 The page cache does the caching. Kafka does not maintain a large in-process record cache. It writes into the OS page cache and lets the kernel flush to disk. Recently written records — the ones consumers are almost always reading, since most consumers run near the tail — are served from RAM without Kafka ever touching the disk on the read path. This also means a broker restart does not cold-start a cache the JVM owns; the page cache is the kernel's and survives process restarts.

🔧 Zero-copy transfer. When serving a fetch request, the broker asks the kernel to send a byte range of a segment file directly to the socket (via sendfile), bypassing a copy into user space and back. The records are already in the exact wire format on disk, so no deserialization or re-encoding happens on the broker.

Without zero-copy:  disk → kernel buffer → app buffer → socket buffer → NIC
With zero-copy:     disk → kernel/page cache ─────────→ NIC

🤔 Did you know? Zero-copy is one concrete reason the broker refuses to inspect your payloads. The moment a broker had to parse a record to filter or transform it, the bytes would need to enter user space and the sendfile path would be unavailable. The "broker sees only bytes" property isn't laziness — it's what makes the fast path fast.

⚠️ One limit of this picture: zero-copy applies to the plaintext path. When the broker must encrypt data for TLS connections, the bytes have to pass through user space, and you pay measurable CPU for it. Broker-side recompression is a separate cost, paid on the write path: if a topic's compression.type differs from what the producer used, the broker decompresses and re-encodes every batch — which is why producing batches already compressed the way the topic stores them is the cheaper arrangement.

Durability, then, is not a property Kafka has; it is a coordinate you pick on a two-axis plane — how many copies exist (replication.factor), and how many you wait for (acks bounded by min.insync.replicas).

✍️ Exercise: accepted or rejected?

A topic has replication.factor=3 and min.insync.replicas=2. For each case, say whether the write is acknowledged, and how many brokers hold the record at that moment.

1. ISR = {L, F1, F2}, producer Acks.All.
2. ISR = {L}, producer Acks.All.
3. ISR = {L}, producer Acks.Leader.
4. ISR = {L, F1}, producer Acks.All. The leader crashes a millisecond after sending the acknowledgement. Is the record lost?

Check your answer

1. Acknowledged once all three hold it — 3 copies.
2. Rejected with NOT_ENOUGH_REPLICAS; nothing is appended.
3. Acknowledged, with 1 copy. min.insync.replicas only constrains acks=all writes. A producer that asks for less is not protected by the broker-side floor — which is why the floor and Acks.All have to be used together.
4. Not lost. The acknowledgement was only sent once F1 had the record — 2 copies, L and F1 — and the new leader is elected from the ISR, which after the crash is just F1. The record is on the new leader.

Producers, Consumers, and the Shape of a Kafka Application

The log is an abstraction until you can see where it touches your code. In a .NET service the day-to-day surface of Kafka reduces to two objects — IProducer<TKey, TValue> and IConsumer<TKey, TValue> from the Confluent.Kafka package, which wraps the native librdkafka C client. Everything else you configure in your service is a knob on one of those two objects (topic settings such as retention and min.insync.replicas live on the cluster). Let's walk the write path, then the read path, at the API-shape level.

The write path: produce, then read the receipt

A producer is a long-lived, thread-safe object. You build it once with a configuration dictionary, keep it for the lifetime of the process, and dispose it on shutdown. Creating one per message is the single most common performance mistake in a first Kafka service, because each producer spins up its own background threads, its own connections to the cluster, and its own metadata cache.

using Confluent.Kafka;

var config = new ProducerConfig
{
    BootstrapServers = "localhost:9092",
    // Human-readable identity that shows up in broker logs and metrics
    ClientId = "orders-api"
};

// Key: string (order id). Value: string (raw JSON, for now).
using var producer = new ProducerBuilder<string, string>(config).Build();

var result = await producer.ProduceAsync(
    "orders",
    new Message<string, string>
    {
        Key = "order-4711",
        Value = """{"orderId":"order-4711","total":42.50}"""
    });

Console.WriteLine($"{result.Topic}[{result.Partition.Value}]@{result.Offset.Value}");
// e.g. orders[2]@10583

The object you get back from ProduceAsync is a delivery report, and it is worth staring at. It contains Topic, Partition, Offset, Timestamp, and a Status. That triple — topic, partition, offset — is the record's address in the cluster forever. Nothing else identifies it. There is no server-assigned message ID, no UUID handed back, no queue entry to look up later. When the offset comes back as 10583, the broker is telling you: your bytes now occupy offset 10583 in that particular log, and they will keep that position for as long as the record exists — until retention removes its segment, or compaction removes it as superseded. (That is an address, not a count — earlier records may already have been removed by retention or compaction.)

That return value also answers a question people ask before they ask anything else: how do I know the write worked? You know because the awaited task completed without throwing. (With Acks.None that proves only that the record left the client: no broker replies, and the returned offset is unset.) If the broker could not satisfy the durability level you asked for, ProduceAsync throws a ProduceException<TKey, TValue> carrying an Error with a code you can branch on. What "satisfied" means is exactly the Acks setting discussed in "Brokers, Replication, and Durability Guarantees" — the delivery report is the client-side face of that broker-side promise.

try
{
    var result = await producer.ProduceAsync("orders", message);
}
catch (ProduceException<string, string> ex)
{
    // ex.Error.Code, e.g. ErrorCode.Local_MsgTimedOut,
    // ErrorCode.NotEnoughReplicas, ErrorCode.TopicAuthorizationFailed
    // ex.DeliveryResult.Status tells you NotPersisted / PossiblyPersisted
    logger.LogError(ex, "Produce failed: {Code}", ex.Error.Code);
    throw;
}

⚠️ Common Mistake: treating a failed produce as "the message definitely wasn't written." The Status on the delivery result distinguishes NotPersisted from PossiblyPersisted — a timeout on the acknowledgement path can mean the record landed but the ack was lost. That ambiguity is precisely why delivery semantics get their own treatment in "Delivery Semantics and Ordering."

Serialization is application code, not broker code

The broker stores opaque bytes. So somewhere between your Order object and the wire, something must produce a byte[] — and in Confluent.Kafka that something is a serializer you attach to the builder. The generic parameters <string, string> above worked without ceremony because the library ships built-in serializers for a handful of primitives (string, int, long, float, double, byte[], Null) and wires them up implicitly. Anything else — including Guid — needs a serializer you supply. The moment your value is a domain type, you supply the conversion yourself.

public sealed class SystemTextJsonSerializer<T> : ISerializer<T>
{
    private static readonly JsonSerializerOptions Options =
        new(JsonSerializerDefaults.Web);

    public byte[] Serialize(T data, SerializationContext context) =>
        JsonSerializer.SerializeToUtf8Bytes(data, Options);
}

public sealed class SystemTextJsonDeserializer<T> : IDeserializer<T>
{
    private static readonly JsonSerializerOptions Options =
        new(JsonSerializerDefaults.Web);

    public T Deserialize(
        ReadOnlySpan<byte> data, bool isNull, SerializationContext context) =>
        isNull ? default! : JsonSerializer.Deserialize<T>(data, Options)!;
}

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

Note the isNull flag on the deserializer: Kafka distinguishes an empty payload from an absent one, and a null value has a specific meaning for compacted topics. Note also SerializationContext, which tells you the topic and whether you are serializing the key or the value — that is the hook Schema Registry–backed serializers use (Schema Registry is a separate service that stores your payload schemas and their versions), since Avro, Protobuf, and JSON Schema serializers register subjects — named schema histories — per topic-and-role.

🎯 Key Principle: the broker will never reject a malformed payload, never validate a field, and never tell you a producer changed shape. Schema compatibility is a contract between your serializer and someone else's deserializer, enforced only by the code on each end (or by a Schema Registry you choose to put in front of them). Making a new field required in the consumer's Order type will break that consumer the moment it replays yesterday's records, which were written without the field — a failure mode revisited in "Common Mental-Model Mistakes and a Working Checklist."

The read path: a pull loop you control

Consuming looks nothing like an event handler or a message-arrived callback. You subscribe, then you call Consume in a loop, and each call returns at most one record. The client fetches batches from the broker in the background and hands them to you one at a time, so the loop is cheap; but the pacing is yours. If your loop is slow, the client stops fetching. Nothing pushes work at you.

var config = new ConsumerConfig
{
    BootstrapServers = "localhost:9092",
    GroupId = "fulfilment-service",
    AutoOffsetReset = AutoOffsetReset.Earliest,
    EnableAutoCommit = false   // we advance the position deliberately
};

using var consumer = new ConsumerBuilder<string, Order>(config)
    .SetValueDeserializer(new SystemTextJsonDeserializer<Order>())
    .Build();

consumer.Subscribe("orders");

try
{
    while (!cancellationToken.IsCancellationRequested)
    {
        var cr = consumer.Consume(cancellationToken); // blocks until a record

        await HandleOrderAsync(cr.Message.Value, cancellationToken);

        // Commits offset+1 synchronously, AFTER the handler succeeded.
        // Only now does our recorded position move past this record.
        consumer.Commit(cr);
    }
}
catch (OperationCanceledException) { /* graceful shutdown */ }
finally
{
    consumer.Close();   // leaves the group immediately instead of waiting
                        // for the session timeout (how long the group waits
                        // on a member that has gone silent) to expire
}

Three details in that loop carry most of the meaning.

🔧 Consume(cancellationToken) blocks. It is a synchronous call on librdkafka's queue, which is why a consumer loop belongs on its own long-running task (a BackgroundService in ASP.NET Core), not interleaved with request handling. There is also a Consume(TimeSpan) overload that returns null on timeout when you want the loop to breathe.

🔧 EnableAutoCommit = false is a deliberate override. In Confluent.Kafka the default for enable.auto.commit is true, meaning positions drift forward on a timer whether or not your handler succeeded. Turning it off is the first step toward controlling restart behavior. The cost is a broker round trip per Commit; the throughput-friendly alternative — leaving auto-commit on but pairing it with EnableAutoOffsetStore = false and calling StoreOffset after the handler — appears in "Common Mental-Model Mistakes." Pick one of the two pairings; mixing them (auto-commit off and StoreOffset with nothing that commits) stores offsets that are never committed.

🔧 consumer.Close() in the finally matters more than it looks. Disposing without closing leaves the group waiting on a session timeout before it redistributes work.

WRITE PATH                          READ PATH

Order object                        fetched byte[] key/value
     ↓ ISerializer                                    ↓ IDeserializer (inside Consume)
  byte[] key/value                             ConsumeResult holding an Order
     ↓ accumulate into batch                          ↓
  send to partition leader                     handle (your code)
     ↓                                                ↓
  broker appends → offset                      advance the position
     ↓
  DeliveryResult back

The symmetry is exact and worth internalising: bytes out, bytes in, with your serialization code as the only translator, and with a receipt at one end and a position decision at the other.

💡 Mental Model: a consumer is a bookmark in a book that other people are still writing at the back. Consume reads the next page; advancing the position moves the bookmark. The book is unchanged either way — the read did not consume anything, in the destructive sense.

Batching and linger: throughput is a setting, not a fate

Here is the part that surprises developers coming from HTTP: ProduceAsync does not send anything to the network. It hands your serialized record to an in-memory queue, and a background thread groups queued records into batches — one batch per destination partition — and sends the batches to each partition's leader broker, several partitions' batches for the same broker travelling in one request.

Three settings govern the grouping:

⚠️ Common Mistake: trusting a linger value from older material. Much JVM-oriented writing says the producer "sends immediately" because the Java client defaulted linger.ms to 0 until Kafka 4.0, which changed it to 5. librdkafka (and therefore Confluent.Kafka) also defaults to 5 ms. Defaults move between versions; if latency or batching matters, set LingerMs explicitly rather than inheriting whichever default your client version happens to have.

Raising LingerMs to, say, 20 buys throughput: bigger batches mean fewer round trips, better compression ratios, and less per-request broker overhead — at the cost of up to 20 ms of added latency for a record that arrives into an empty batch. Lowering it toward zero minimises latency at the cost of many small requests. Neither is correct in the abstract; a click-stream ingest job and a synchronous payment authorisation want opposite ends of that dial.

💡 Real-World Example: a service producing 50 000 small events per second with LingerMs = 0 may issue tens of thousands of produce requests per second. Raising linger to 10 ms lets those events coalesce into far fewer, larger requests — the same bytes, dramatically less request overhead, and the added latency is invisible to a batch pipeline.

Because records sit in memory, shutdown requires a flush. Disposing the producer does not wait for them: librdkafka's documented termination sequence is flush first, then destroy, and records still queued or in flight at destroy time are dropped. Call Flush with a bound you chose, short enough to finish before the orchestrator's grace period ends:

// Block until the in-memory queue drains, or the timeout elapses.
var stillQueued = producer.Flush(TimeSpan.FromSeconds(10));
if (stillQueued > 0)
    logger.LogError("{Count} messages never delivered", stillQueued);

🤔 Did you know? There is a non-awaiting overload, producer.Produce(topic, message, handler), that returns immediately and invokes your callback on the delivery report later — unlike Acks.None, the broker still acknowledges every record; you just don't wait for it. It is substantially faster than awaiting each ProduceAsync, because awaiting per message serialises your code against network round trips. The trade is that error handling moves into the callback, so you need somewhere for failures to go — and, because nothing awaits the delivery, a Flush before shutdown is the only thing standing between the queue and the drop described above.

✍️ Exercise: spot the lost writes

A teammate's endpoint does this on every HTTP request:

app.MapPost("/orders", (Order order) =>
{
    using var producer = new ProducerBuilder<string, string>(config).Build();
    producer.Produce("orders", new Message<string, string>
    {
        Key = order.Id,
        Value = JsonSerializer.Serialize(order)
    });
    return Results.Accepted();
});

Name two problems, and say which one loses data.

Check your answer

1. Lost writes. Produce only enqueues the record. The using disposes the producer as the handler returns, and records still queued or in flight at that point are dropped — the client got 202 Accepted for an order that may never reach Kafka. Nothing reports the failure, because no delivery handler was passed.
2. A producer per request. Each instance starts its own background threads, broker connections and metadata cache, so the endpoint pays that setup on every call. Register one producer as a singleton, dispose it (after Flush) on shutdown, and either await ProduceAsync or pass a delivery handler so failures are visible.

The three decisions this shape leaves open

The code above deliberately left three questions hanging, and each is a knob you can now locate precisely in the API surface:

🔑 Message.Key — the string "order-4711" decided which partition the record landed in, which is why the delivery report named partition 2 and not some other partition. Key-to-partition placement, and what happens when the key is null, belongs to the "Topics, Partitions & Keys" lesson.

👥 ConsumerConfig.GroupId — the value "fulfilment-service" decided how work is split when you run three instances of this service. Group membership and rebalancing (reshuffling which instance reads which partition as instances join or leave) belong to the "Consumer Groups & Offsets" lesson.

📍 Commit and AutoOffsetReset — these two decided where the loop resumes. After a crash it resumes from the last committed position; AutoOffsetReset only applies when the group has no committed position for a partition (a brand-new group) or that position has already been removed by retention. Commit timing, and the at-least-once versus at-most-once consequence of committing before or after processing, belong to the "Consumer Groups & Offsets" lesson and to "Delivery Semantics and Ordering" below.

In short: the key decides where a record goes, the group decides who processes it, and the committed offset decides where processing resumes. Three fields, three orthogonal decisions.

Everything else you will build — retries, dead-letter topics (side topics where records that cannot be processed are parked), parallel handlers, transactional outboxes — is composed from the two loops on this page.

Delivery Semantics and Ordering: What Kafka Actually Promises

Most production incidents traced back to Kafka aren't caused by Kafka failing to do what it promised. They're caused by a design that assumed a promise Kafka never made. Two assumptions come up again and again: that a topic delivers records in the order they were produced, and that "exactly-once" means your handler runs exactly once. Both are false in ways that matter, and both are cheap to design around once you know the real boundary.

Ordering Stops at the Edge of a Single Log

Kafka guarantees that records appended to one partition are read back in exactly the order they were appended, with monotonically increasing offsets. That's the whole guarantee. There is no cross-partition ordering, no cluster-wide clock, no merge step that reconciles partitions on the way out.

Topic "orders" — three of its partitions shown

P0 offsets:  0:order-17   1:order-42   2:order-17    ← strictly ordered
P1 offsets:  0:order-08   1:order-91                ← strictly ordered
P2 offsets:  0:order-42                             ← strictly ordered

Across P0 / P1 / P2: no defined order whatsoever.
Two consumers can observe P0[2] before or after P1[0].

Notice order-17 appears twice in P0 (two events for the same order, correctly sequenced) while order-42 appears in both P0 and P2 — the same business entity landed in two logs, and its two events now have no relative order at all. A stable key and a fixed partition count would have prevented that (with one .NET-specific caveat: librdkafka's default partitioner hashes keys with CRC32, while the Java client uses murmur2, so a .NET and a Java producer writing the same key land in different partitions unless you set Partitioner = Partitioner.Murmur2Random in Confluent.Kafka), so the split usually tells you something went wrong: the events were produced with different keys (or none), by producers with different partitioners, or the topic was repartitioned between them. Keeping related events in one log is the job of the message key, which gets its own lesson; here the point is the consequence: ordering is a property you buy by co-locating records, not a property the topic hands you.

❌ Wrong thinking: "I'll publish OrderCreated, OrderPaid, and OrderShipped to the same topic, so consumers see them in order."

✅ Correct thinking: "Those three events must share a partition — same key, same topic — or my consumer must tolerate arriving out of order (version numbers, state machine that ignores stale transitions, or a reconciliation pass)."

⚠️ Common Mistake: Even within one partition, a non-idempotent producer with more than one in-flight request per connection can reorder on retry. If batch 5 fails and batch 6 succeeds, the retry of 5 lands after 6. In Confluent.Kafka this is governed by MaxInFlight (max.in.flight.requests.per.connection); with retries enabled and a value above 1, in-partition ordering is not guaranteed until you turn on idempotence. Both of those are librdkafka's defaults (effectively unlimited retries and up to 1,000,000 in-flight requests), so a default, non-idempotent Confluent.Kafka producer is in exactly this situation.

🎯 Key Principle: Ordering scope = one partition, one producer, ordering-preserving producer settings. Anything wider is your application's problem, not the broker's.

The Three Delivery Semantics, and Who Pays for Each

At-most-once means a record may be lost but never processed twice. You get it by advancing your read position before processing. If the process dies mid-handler, the record is skipped on restart. This is the right choice for metrics samples or cache warm-ups, and catastrophic for payments.

At-least-once means a record is never lost but may be processed more than once. You get it by advancing your position after processing succeeds. Crash between "processed" and "position advanced" and the record replays. This is the default posture of virtually every well-behaved consumer.

Effectively-once means duplicates may still be delivered, but they have no observable effect. Note the wording: nobody suppresses the redelivery of a record to your handler across a crash. The duplicate is neutralized either by your write being idempotent, or — when the write goes to another Kafka topic — by Kafka transactions making that write and the offset commit atomic.

The consumer column is what actually selects the semantic; the producer column is the setting that keeps the write side from adding its own losses or duplicates on top.

Idempotent Producers: Sequence Numbers Do the Deduplication

A producer retry is not optional — a lost acknowledgement is indistinguishable from a lost write, so the client must resend. An idempotent producer makes that resend safe. On initialization the producer is assigned a producer ID (PID) and an epoch (a generation number for that ID), and it stamps every record batch with a per-partition sequence number. The partition leader tracks the last sequence it accepted per PID and recognizes a repeat as a duplicate and acknowledges it as success without appending a second copy.

Producer (PID 4711, epoch 0) → partition P0
  ↓ send seq=5      → leader appends seq=5
  ↓ ack lost in the network (producer sees a timeout)
  ↓ retry  seq=5    → leader: "seq 5 already applied"
                       → returns success, appends nothing
  ↓ send seq=6      → appended normally

using Confluent.Kafka;

var config = new ProducerConfig
{
    BootstrapServers = "broker-1:9092,broker-2:9092",

    // In Confluent.Kafka (librdkafka) this is NOT enabled by default —
    // unlike the JVM client, which turns it on for you. Set it explicitly.
    EnableIdempotence = true,

    // With EnableIdempotence=true, librdkafka enforces the settings that make
    // the guarantee possible: Acks=All, bounded MaxInFlight, and retries.
    // Setting them yourself to conflicting values is a startup error.
    MessageSendMaxRetries = int.MaxValue,
    MessageTimeoutMs = 300_000
};

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

var result = await producer.ProduceAsync("orders",
    new Message<string, string> { Key = "order-17", Value = payloadJson });

// TopicPartitionOffset tells you exactly which log position accepted the record.
Console.WriteLine($"{result.TopicPartitionOffset} status={result.Status}");

⚠️ The guarantee is bounded by the producer session. Dedup state lives per PID, so if your service restarts and re-sends a record it isn't sure about, the new producer instance carries a new PID and the broker cannot recognize the duplicate. Even without a restart, a record your code produces again is a new record with a new sequence number. Idempotence protects against client-library retries, not against your own application-level "send it again" logic.

🤔 Did you know? Because the broker validates sequence numbers per PID and partition, an idempotent producer cannot silently reorder or duplicate records on retry. It can still leave a gap: if a batch fails permanently, or a record exceeds MessageTimeoutMs, that record's delivery report carries the error and later records are still written. librdkafka offers an experimental enable.gapless.guarantee setting that turns such gaps into a fatal producer error instead; it is off by default. Either way, the delivery report is where you learn a record was not written — so check it.

The Pragmatic Default: At-Least-Once Plus an Idempotent Consumer

For the large majority of .NET services that consume Kafka and write to a relational store, the winning combination is an idempotent producer, at-least-once consumption, and a write that can absorb duplicates. Three patterns cover most cases:

🔧 Upsert by business key — MERGE/ON CONFLICT on the natural identifier, so replaying the same event twice converges to the same row.

🔧 Conditional write — include the source version or offset in the WHERE clause so a stale or repeated event is a no-op.

🔧 Dedupe table — record the event's unique ID in a table with a unique constraint inside the same transaction as the effect; a duplicate insert fails and you skip the work.

// Idempotent handler: conditional upsert keyed on the business key,
// guarded by a monotonically increasing version from the event itself.
// (PostgreSQL syntax; on SQL Server this is a MERGE with the same WHERE guard.)
await using var tx = await conn.BeginTransactionAsync(ct);

var rows = await conn.ExecuteAsync("""
    INSERT INTO order_state (order_id, status, version)
    VALUES (@OrderId, @Status, @Version)
    ON CONFLICT (order_id) DO UPDATE
       SET status  = EXCLUDED.status,
           version = EXCLUDED.version
     WHERE order_state.version < EXCLUDED.version;
    """,
    new { evt.OrderId, evt.Status, evt.Version }, tx);

// rows == 0 means a duplicate or an out-of-order replay: correct outcome, no work done.
await tx.CommitAsync(ct);

// Position advances only after the write commits → at-least-once.
consumer.Commit(consumeResult);

That single WHERE version < clause buys you both duplicate tolerance and out-of-order tolerance across a partition change — which is why it is worth more than any broker setting. (This snippet assumes the store commits durably; if your handler also calls an external non-idempotent API, you need a dedupe table around that call too.)

🧠 Mnemonic: DIP — Duplicates arrive, Idempotent write absorbs them, Position advances last.

✍️ Exercise: where can it crash?

This at-least-once handler writes to SQL Server with a plain INSERT (no upsert, no dedupe table):

var cr = consumer.Consume(ct);                 // ①
await _db.InsertPaymentAsync(cr.Message.Value); // ② commits its own DB transaction
consumer.Commit(cr);                            // ③

For a crash at each point, say what happens to that record after restart and whether the payment ends up in the database zero, one or two times:

• (a) during ②, before the database transaction commits;
• (b) after ② commits, before ③;
• (c) after ③.

Then: what single change turns the risky case into a harmless one?

Check your answer

• (a) The record is read again (its offset was never committed) and the insert runs for the first time → once.
• (b) The record is read again and the insert runs a second time → twice. This is the duplicate that at-least-once allows.
• (c) The position is past the record; it is not read again → once.

The fix is on the write side, not in Kafka: make ② idempotent — an upsert keyed on the payment ID, or a dedupe table with a unique constraint written in the same database transaction. Then case (b) converges to one row. Moving Commit before ② would not help; it would turn case (a) into a lost payment (at-most-once).

Transactions and read_committed: Real, and Genuinely Expensive

Kafka transactions let a producer write to several topic-partitions and commit its consumed offsets as one atomic unit. Assign a stable TransactionalId, and the cluster's transaction coordinator fences out any older instance using the same ID — so a zombie pod that comes back to life cannot commit.

// Assumes `cr` is the ConsumeResult currently being processed, obtained from
// consumer.Consume(...) on a consumer subscribed with the settings below.
var txConfig = new ProducerConfig
{
    BootstrapServers = "broker-1:9092",
    TransactionalId = "enricher-v1",   // stable per logical instance; enables fencing
    EnableIdempotence = true
};

using var producer = new ProducerBuilder<string, string>(txConfig).Build();
producer.InitTransactions(TimeSpan.FromSeconds(30));

// Consumer must NOT auto-commit: the transaction owns the offsets.
var consumer = new ConsumerBuilder<string, string>(new ConsumerConfig
{
    BootstrapServers = "broker-1:9092",
    GroupId = "enricher",
    EnableAutoCommit = false
}).Build();

producer.BeginTransaction();
try
{
    producer.Produce("orders-enriched", new Message<string, string> { Key = k, Value = v });
    producer.Produce("orders-audit",    new Message<string, string> { Key = k, Value = a });

    producer.SendOffsetsToTransaction(
        new[] { new TopicPartitionOffset(cr.TopicPartition, cr.Offset + 1) },
        consumer.ConsumerGroupMetadata,
        TimeSpan.FromSeconds(30));

    producer.CommitTransaction();   // both writes + the offset become visible together
}
catch (KafkaTxnRequiresAbortException)
{
    // Recoverable: roll back, rewind the consumer to the last committed
    // offsets, and retry the batch.
    producer.AbortTransaction();
}
catch (KafkaException ex) when (ex.Error.IsFatal)
{
    // NOT abortable — the producer instance is dead. Dispose it and build a
    // new one with the same TransactionalId; calling AbortTransaction here
    // would only throw again.
    throw;
}

Not every failure is abortable, which is why the two catches differ. KafkaTxnRequiresAbortException is the cluster telling you to roll back and retry; a fatal KafkaException means the producer can no longer participate in transactions at all and must be rebuilt. Treating both the same way — abort and continue — leaves you looping on a producer that will never commit again.

Downstream, only a consumer with IsolationLevel = IsolationLevel.ReadCommitted will skip aborted records and refuse to read past the last stable offset (the point where the oldest still-open transaction begins). ⚠️ Confluent.Kafka's underlying isolation.level default is read_committed, which differs from the JVM client's read_uncommitted default — if you are porting configuration between the two, set it explicitly rather than assuming parity.

The cost is not theoretical. Every transaction adds coordinator round-trips, aborted records still occupy log space, and a read_committed consumer cannot advance past an open transaction — so one slow producer stalls readers, converting a producer hiccup into consumer lag. Transactions also stop at Kafka's boundary: they make Kafka writes plus Kafka offsets atomic, not your SQL insert. A consume → write-to-database → commit-offset flow gains nothing from them, which is precisely why the idempotent-consumer pattern above remains the default for most .NET services, and transactions are reserved for read-process-write pipelines that stay entirely inside Kafka.

Common Mental-Model Mistakes and a Working Checklist

Kafka incidents that trace back to a wrong mental model usually leave a matching signature in logs or metrics. If you can recognize the symptom, you can diagnose the misconception behind it in minutes instead of days. Here are the four that recur most, each with its correction and its observable fingerprint.

Mistake 1: Treating Kafka as a queue ⚠️

This one comes up constantly, because Kafka sits in the same architectural slot that RabbitMQ or Azure Service Bus used to occupy, and the vocabulary overlaps ("consumer", "message", "acknowledge"). The words are the same; the semantics are not.

❌ Wrong thinking: "I consumed the message, so it's gone. The broker knows I'm working on this message; if I crash before acknowledging it, the broker will notice and redeliver it."

✅ Correct thinking: "I read a copy at offset N. The record is still in the log until retention removes it. The broker knows only my group's committed position. If I crash, whoever takes over my partition resumes from that position — so this record is re-read only if I had not yet committed past it, and every uncommitted record after it is re-read too."

The practical fallout: there is no broker-side per-message state machine. Kafka does not track "delivered but unacked", it does not have a visibility timeout, it does not have a per-message dead-letter hook, and it will not requeue a single failed record while letting the rest of the partition flow past it. The commit you make is a position, not an acknowledgement of one message.

(Kafka 4.2 made share groups — "Queues for Kafka" (KIP-932) — production-ready: a separate kind of group with per-record acknowledgement, delivery counts and redelivery, built on top of the same log. Everything in this lesson describes ordinary consumer groups, which is what ConsumerBuilder and GroupId give you. Check your client's support before planning around share groups.)

Symptom to recognize: a consumer that "loses" work after a crash even though nothing errored — because it committed offsets before processing finished. Or the inverse: after a rebalance you see the same records processed again, followed by duplicate side effects downstream. Another tell is a team building a Nack() helper and discovering there is nothing to call.

💡 Real-World Example: A team ports a Service Bus worker to Kafka and keeps the old shape — on exception, log it and move on without completing the message, relying on Service Bus to redeliver it once the lock expires. Under Kafka, nothing returns. The offset advances at the next commit and the record is silently skipped. The metric that reveals it: business-level counts (orders processed) drift below produced counts while consumer lag sits comfortably at zero. Lag being zero is not evidence that work was done — only that the cursor moved.

Mistake 2: Treating Kafka as a database

The log is durable, replicated, and can hold months of data. That combination tempts people into treating a topic as a database they can query. The log can be the authoritative record of what happened — but it is not a query engine. The broker never inspects the payload — as covered in "The Log Is the Data Structure", records are opaque bytes — so it cannot filter, index, or join on anything inside them.

The only random access Kafka offers is by offset or by timestamp within a partition — you can seek to a position, then read forward. That is a scan, not an index. A topic with a hundred million records and no consumer-side projection means answering "what is the state of order 4711?" costs a full traversal of the partition that key landed on.

Symptom to recognize: a service that starts up and reads the entire topic from the beginning on every deploy, with startup times that grow linearly with topic size. Or a support tool whose "lookup by ID" endpoint times out. Or brokers running out of disk because someone set an effectively infinite retention as a substitute for an archive.

🎯 Key Principle: Kafka is the transport and the ordered history. The query-answering copy is something you build by consuming, and it lives in SQL Server, Postgres, Redis, or a search index — wherever your access pattern is cheap.

💡 Mental Model: A topic is a bank statement, not a bank balance. The statement is the authoritative record of what happened; the balance is a derived thing you compute and keep somewhere fast to read.

Mistake 3: Blocking work and poison messages inside the consume loop

A Kafka partition is a strictly ordered single-file line of work, and within your group only your consume loop is draining it. One loop usually serves every partition assigned to that instance, so anything slow you do inside it holds back every record behind it — in the same partition and in all the other partitions the instance owns. This is the anti-pattern:

// ❌ ANTI-PATTERN: slow I/O and unbounded retry inside the consume loop
while (!cancellationToken.IsCancellationRequested)
{
    var result = consumer.Consume(cancellationToken);

    // Retries forever on a poison record: everything behind it stalls.
    while (true)
    {
        try
        {
            var body = new StringContent(
                result.Message.Value, Encoding.UTF8, "application/json");
            var response = await _httpClient.PostAsync(url, body, cancellationToken);
            response.EnsureSuccessStatusCode(); // a 4xx/5xx now throws HttpRequestException
            break;
        }
        catch (HttpRequestException ex) // network failures AND non-success status codes
        {
            _logger.LogWarning(ex, "Retrying...");
            await Task.Delay(TimeSpan.FromSeconds(5), cancellationToken);
        }
    }

    consumer.Commit(result);
}

Two failures compound here. First, if the downstream endpoint returns a permanent 400 for one malformed record, this loop never exits — none of the instance's partitions advance, and lag climbs on all of them. Second, Consume is not being called during the retry, and in Confluent.Kafka the consumer must return to Consume within max.poll.interval.ms (default 300000 ms — five minutes) or the client treats itself as failed, leaves the group, and its partitions are rebalanced away. When the member finally wakes up, its commit fails because it no longer owns the partition.

Symptom to recognize: Application maximum poll interval exceeded in librdkafka logs, repeated Rebalance events with no deployment happening, a Commit throwing KafkaException with ErrorCode.UnknownMemberId or IllegalGeneration, and one partition whose committed offset never moves while its lag climbs steadily. After the rebalance the stuck instance's healthy partitions recover on other members, but the partition holding the poison record stalls whichever member picks it up next, holding back that member's other partitions until it is evicted in turn — so one committed offset that never moves while the rest drain in fits and starts, with rebalances recurring, is the poison-record signature.

The correction is a bounded retry plus an explicit escape hatch: try a small, finite number of times for transient errors, and for anything that will never succeed, route the record to a dead-letter topic (often called a DLQ, after the queue world's dead-letter queue) and move on.

// Consumer config for this loop — the auto-commit pairing that makes
// StoreOffset meaningful:
//   EnableAutoCommit      = true    (the client commits on a timer)
//   EnableAutoOffsetStore = false   (but only offsets WE store)

// ✅ Bounded retries, poison records routed to a DLQ, loop keeps moving
while (!cancellationToken.IsCancellationRequested)
{
    var result = consumer.Consume(cancellationToken);

    try
    {
        await ProcessWithBoundedRetryAsync(result.Message, cancellationToken);
    }
    catch (Exception ex) when (ex is not OperationCanceledException)
    {
        // Permanently failing record: preserve it, don't block the partition.
        // (Shutdown cancellation is excluded: that record is healthy and must be re-read.)
        await dlqProducer.ProduceAsync("orders.dlq", new Message<string, string>
        {
            Key = result.Message.Key,
            Value = result.Message.Value,
            Headers = new Headers
            {
                { "x-original-topic", Encoding.UTF8.GetBytes(result.Topic) },
                { "x-original-partition", Encoding.UTF8.GetBytes(result.Partition.Value.ToString()) },
                { "x-original-offset", Encoding.UTF8.GetBytes(result.Offset.Value.ToString()) },
                { "x-error", Encoding.UTF8.GetBytes(ex.GetType().Name) }
            }
        }, cancellationToken);

        _logger.LogError(ex, "Dead-lettered {Topic}/{Partition}@{Offset}",
            result.Topic, result.Partition, result.Offset);
    }

    // Store the position after the outcome is decided (success or dead-lettered).
    consumer.StoreOffset(result);
}

Both consumer and DLQ producer are typed <string, string> here so the payload passes through untouched. Records that fail to deserialize never reach this try at all: Consume itself throws a ConsumeException. Its ConsumerRecord property still carries the raw key and value bytes, so a separate catch (ConsumeException) around Consume can forward them to the dead-letter topic with a byte[] producer.

Note the headers: partition and offset go with the record, so a replay tool can reconstruct exactly where it came from. StoreOffset marks the position for the background auto-committer — in Confluent.Kafka, enable.auto.commit defaults to true, and pairing it with enable.auto.offset.store=false gives you the useful combination of you deciding which offsets are eligible while the client handles commit timing. Offset-commit strategy in depth belongs to the "Consumer & Manual Commits" lesson; the point here is that the decision to advance must come after the outcome is known.

💡 Pro Tip: If a single record legitimately needs ten minutes of work — longer than the five-minute default — don't stretch max.poll.interval.ms to cover it. Hand the work to a bounded channel, Pause that partition while it runs, and keep calling Consume so the member stays in the group; Resume it and advance the position once the work is done. (Adding partitions does not help: one ten-minute record still blocks whichever loop holds it, and on a keyed topic it changes which partition each key maps to.) Raising the poll interval just delays how long it takes the group to notice a genuinely stuck member.

Mistake 4: Changing payload shape without a compatibility plan

The broker stores bytes and validates nothing, so a breaking schema change produces no error at produce time. It produces a deserialization exception in someone else's service, possibly weeks later.

What makes Kafka distinctive here is retention. In a request/response system, once you deploy the new client and server, old payloads are gone. In Kafka, a record written months ago is still in the log, and any consumer resetting to earliest — a new service, a rebuilt projection, a bug-fix replay — will read the old shape with today's code. Both directions matter: new code reading old records, and old code (a consumer you forgot to redeploy) reading new records.

With plain System.Text.Json and no registry, you enforce this by discipline: never rename, never retype, add fields as nullable with defaults, and treat unknown enum values as a handled case rather than an exception. That is workable for a small system and it is where most teams start.

⚠️ The built-in JsonStringEnumConverter does not give you that last one — it throws a JsonException on a member it doesn't recognize, which is exactly the failure you were trying to avoid when a newer producer emits a value your build has never seen. Tolerating unknown members needs a converter that falls back:

// Falls back to the zero member instead of throwing on an unrecognized name.
public sealed class TolerantEnumConverter<T> : JsonConverter<T> where T : struct, Enum
{
    public override T Read(ref Utf8JsonReader reader, Type _, JsonSerializerOptions __) =>
        reader.TokenType == JsonTokenType.String
        && Enum.TryParse<T>(reader.GetString(), ignoreCase: true, out var value)
        && Enum.IsDefined(value)   // TryParse also accepts undefined numbers like "7"
            ? value
            : default;   // member 0 — name it Unknown

    public override void Write(Utf8JsonWriter writer, T value, JsonSerializerOptions _) =>
        writer.WriteStringValue(value.ToString());
}

// Version-tolerant contract: safe to evolve, safe to replay old records.
public sealed record OrderPlaced
{
    public required string OrderId { get; init; }
    public required decimal Total { get; init; }

    // Added later — nullable with a default, so records written before
    // this field existed still deserialize cleanly.
    public string? PromotionCode { get; init; }

    // A future "Kiosk" value deserializes to Unknown rather than throwing.
    [JsonConverter(typeof(TolerantEnumConverter<OrderChannel>))]
    public OrderChannel Channel { get; init; } = OrderChannel.Unknown;
}

public enum OrderChannel { Unknown = 0, Web = 1, Mobile = 2 }

The scalable answer is a schema registry with a compatibility mode enforced at registration time, so an incompatible schema is rejected when it is registered — at the producer's first send by default, since the serializers auto-register, or in CI if you register schemas there — rather than discovered by a 3 a.m. page. Confluent.Kafka has companion serializer packages for Avro, Protobuf, and JSON Schema that register the producer's schema with Schema Registry — which accepts or rejects it under the subject's compatibility mode — and embed the schema ID in each record.

Symptom to recognize: JsonException: The JSON value could not be converted or ConsumeException with ErrorCode.Local_ValueDeserialization appearing on one consumer group while others are fine — that is the group that reset its offsets and is now replaying pre-change history. A deserialization error on the newest offsets instead points the other way: a consumer that wasn't redeployed.

🤔 Did you know? Deserialization failures in Confluent.Kafka surface as a ConsumeException from Consume() — but by then the client has already moved past that record. If you catch, log, and loop, the next Consume returns the next record, and once a later offset is stored or committed the failed one is behind your position for good. The failure mode is the opposite of a poison-message stall: data silently skipped. Handle ConsumeException by dead-lettering ex.ConsumerRecord's raw bytes before carrying on.

The five facts every design decision traces back to

When a Kafka design question comes up — "should this be one topic or three?", "can we guarantee this runs once?", "why did we get duplicates?" — the answer is almost always derivable from five facts. Run the question against each of them.

1. LOG SEMANTICS
   ↓  Append-only, immutable, read by position. Reads never remove.
2. REPLICATION & ACKS
   ↓  Durability is copies (replication.factor), a producer choice
      (acks) and a broker floor (min.insync.replicas). A fetched
      record is on every in-sync replica, which may be the leader alone.
3. ORDERING SCOPE
   ↓  Ordering exists inside one partition only (per producer, with
      ordering-preserving settings). Never across partitions.
4. DELIVERY SEMANTICS
   ↓  At-least-once is the practical default; idempotent processing
      is what makes it correct.
5. RETENTION
   ↓  Records leave via time, size, or compaction (or a deliberate
      admin delete) — never because someone read them.

Fact 3 is often the first one a design gets wrong: ordering does not extend across partitions.

Applied to a real question: "Our fraud service must see every transaction for a customer in order — how do we build it?" Fact 3 says ordering only exists within a partition, so all events for one customer must land in one partition; fact 1 says other services can read the same log independently without interfering; fact 4 says the fraud service will occasionally see a record twice, so its writes need to be idempotent; fact 5 says if it goes down over a long weekend, whether it can catch up depends on retention, not on the queue depth. Four of the five facts answered the design in one pass.

✍️ Exercise: apply the five facts yourself

An inventory service consumes OrderPlaced events. For each event it must decrement stock in PostgreSQL once and send the customer one confirmation email through a third-party API. A colleague proposes "just turn on Kafka transactions — they give exactly-once." Using the five facts, say why that proposal does not solve the problem and sketch what does.

Check your answer

• Fact 4 (delivery): the consumer is at-least-once, so some OrderPlaced events will be handled twice after a crash or rebalance. Kafka transactions only make Kafka writes plus Kafka offsets atomic; the PostgreSQL update and the email are outside Kafka, so a transaction does not stop them happening twice.
• Stock: make the write idempotent — record the event ID in a dedupe table with a unique constraint in the same database transaction as the decrement. A replayed event hits the constraint and does nothing.
• Email: the third-party call cannot join that database transaction. Record "email due for event X" in the same transaction (an outbox row) and have a sender mark it done; if the email API supports an idempotency key, pass the event ID so a retry after a crash is not sent twice.
• Fact 3 (ordering): events for the same SKU or order only arrive in order if they share a key. Decrements commute, so strict ordering matters less here than idempotence.
• Fact 5 (retention): if the service is down longer than the topic's retention, events are gone before it can catch up — size retention for your worst realistic outage.

⚠️ Remember: One of the most expensive assumptions is that a commit means "this message is done" rather than "my cursor is here now." Many duplicate-processing and silent-data-loss incidents in Kafka consumers are that one substitution playing out.

Three things worth doing next with this model in hand: add a dashboard panel for per-partition lag (not just topic-level average — poison records hide inside averages); audit your consume loops for any await that can block unboundedly and put a timeout on it; and write down the compatibility rule your team will follow for payload changes before you need it. The next lessons build on these five facts — keys and partition assignment, consumer groups, and offset management each build on the list above.