Debugging a Kafka Protocol Implementation: When 11 Messages Become 1

You know that feeling when everything seems to work—connection establishes, data flows, no errors anywhere—but something is silently, catastrophically wrong? This is the story of tracking down exactly that kind of bug.

TL;DR: Implementing Kafka protocol compatibility taught me that eight bytes in the wrong place can cause complete data loss with zero errors. The producer acknowledged 11 messages. The consumer received 1. Both sides thought they were done. This post walks through how I found and fixed the bug—and why this debugging session taught me more about Kafka internals than months of using it as a black box.

Why implement a protocol yourself?

Before diving into the bug, let me address the obvious question: why build a Kafka-compatible server when Kafka already exists?

Three reasons:

1. You don't truly understand a protocol until you implement it. I've used Kafka for years—configuring producers, tuning consumers, debugging lag issues. But I never really understood why it worked the way it did until I had to make something speak its language.

2. Edge cases become visceral. Reading that "the broker must update baseOffset" is one thing. Watching 10 messages vanish because you forgot to do it? That sticks with you.

3. It changes how you use the real thing. After this experience, I now know exactly what's happening when a Kafka consumer "loses" its place, or why certain configurations matter. The mental model is no longer abstract.

If you want to truly understand a distributed system, try implementing its protocol. You'll learn more from your first bug than from a dozen architecture diagrams.

Now, let's see what happens when you miss a critical detail.

I was working on Streamline, a Kafka-compatible streaming server I've been building. The goal is straightforward: existing Kafka clients should work unchanged against Streamline. We're not trying to be a full Kafka replacement—think of it as "Redis for streaming"—but we do need to speak the Kafka wire protocol correctly.

Everything seemed to be working. Unit tests passed. The server started. Clients connected. Life was good.

Then I ran a simple end-to-end test:

# Producer sends 11 messages
$ echo -e "msg1\nmsg2\nmsg3\nmsg4\nmsg5\nmsg6\nmsg7\nmsg8\nmsg9\nmsg10\nmsg11" | \
    kafka-console-producer --bootstrap-server localhost:9092 --topic test-topic
>>>>>>>>>>>
 
# Consumer reads... 1 message?!
$ kafka-console-consumer --bootstrap-server localhost:9092 \
    --topic test-topic --from-beginning
msg1
Processed a total of 1 messages

Eleven arrows (one per message successfully acknowledged). One message received. What the hell?

Why this bug was particularly evil

Here's what made this frustrating:

This is the worst category of bug: silent data loss with no indication anything is wrong. The producer got acknowledgments. The consumer exited after "processing" its single message. Everyone was happy except the 10 messages that vanished into the void.

Kafka's record batch format: the background you need

Before I explain what went wrong, you need to understand how Kafka structures messages on the wire. This isn't academic—it's the key to everything.

Kafka doesn't send individual messages

When you call producer.send(), Kafka doesn't immediately send that message. Messages are batched together for efficiency. The kafka-console-producer with 11 newline-separated messages sends them as a single Record Batch.

From Apache Kafka's official documentation:

"Messages are written to Kafka in batches... The on-disk format of a RecordBatch is designed to be the same as the in-memory format for zero-copy sending to consumers."

The byte layout (this is where it gets good)

Looking at Kafka 4.3.1's DefaultRecordBatch.java, here's the exact byte layout of a Record Batch header:

The header consumes 61 bytes before record data begins. The field list and offsets come from the Kafka protocol specification; the visualization keeps that layout available without forcing the reader to parse a monospaced table.

The critical field: baseOffset

Here's what the KIP-98 specification says about baseOffset:

"The offset provided in the message set header represents the offset of the first message in the set."

And from the Kafka Protocol Guide:

"The broker will update the baseOffset field in each RecordBatch to the actual offset assigned in the log."

Let me emphasize: the broker MUST update the baseOffset. The producer sends batches with baseOffset = 0 (relative offset). The broker must rewrite those first 8 bytes to contain the actual log offset where the batch is being stored.

How consumers calculate record offsets

Each record within a batch has an offsetDelta field (a varint, not fixed-size). The actual offset of each record is:

record_offset = baseOffset + offsetDelta

So for a batch with 11 records:

• Record 0: baseOffset + 0
• Record 1: baseOffset + 1
• ...
• Record 10: baseOffset + 10

The batch header also contains lastOffsetDelta (at byte offset 23), which for 11 records would be 10. This lets consumers calculate the next expected offset:

next_expected_offset = baseOffset + lastOffsetDelta + 1

You might be wondering: "Why does Kafka bother with this complexity?" Because it enables zero-copy transfers and efficient offset tracking without parsing every individual record. Pretty clever, right?

Tracing the bug

Armed with this knowledge, let's trace what was happening.

Step 1: Verify data is arriving

First, I needed to confirm the producer was actually sending 11 messages:

debug!(
    topic = %topic_name,
    partition = partition_index,
    records_len = batch_bytes.len(),
    "Processing produce request"
);

Result: ✅ All 11 messages arrived in a single batch.

Step 2: Check storage

Looking at our storage code:

// Original produce handler
let value = Bytes::copy_from_slice(records);  // Raw batch bytes
match self.topic_manager.append(&topic_name, partition_index, None, value) {
    Ok(offset) => {
        (offset, NONE)  // Returns offset 0
    }
    // ...
}

And in Partition::append():

pub fn append(&mut self, key: Option<Bytes>, value: Bytes) -> Result<i64> {
    // Get next offset - always increments by 1
    let offset = self.next_offset.fetch_add(1, Ordering::SeqCst);
    // ... store record at 'offset' ...
    Ok(offset)
}

First problem found: We increment offset by 1 regardless of how many records are in the batch.

Step 3: Understand the consumer's perspective

Here's the sequence from the consumer's viewpoint:

1. Consumer connects, requests metadata for test-topic
2. Consumer sends FetchRequest with fetch_offset = 0
3. Server returns batch containing records 0-10
4. Consumer reads batch, sees baseOffset = 0, lastOffsetDelta = 10
5. Consumer calculates: "I received offsets 0-10, next I need offset 11"
6. Consumer sends FetchRequest with fetch_offset = 11
7. Server looks for data at offset >= 11
8. Server's internal offset counter is at 1 (we only had one append() call)
9. Server finds nothing at offset 11
10. Server returns empty FetchResponse
11. Consumer thinks it's caught up, exits

And just like that, 10 messages vanish.

Step 4: But wait, there's more

Even if we fixed the offset counting, there's a second problem. Remember: the broker must update baseOffset.

Our code was storing the raw batch bytes without modification. The producer sent baseOffset = 0 in the batch header, and we stored those exact bytes. For the first batch, this happened to work because we also assigned offset 0. But for any subsequent batch? Disaster.

Correct behavior:

Batch 1 arrives: producer's baseOffset = 0, lastOffsetDelta = 10
Broker assigns:  baseOffset = 0, stores at offset 0
                 Updates header bytes [0..8] to contain 0 (big-endian)
                 Increments next_offset by 11

Batch 2 arrives: producer's baseOffset = 0, lastOffsetDelta = 2
Broker assigns:  baseOffset = 11, stores at offset 11
                 Updates header bytes [0..8] to contain 11 (big-endian)
                 Increments next_offset by 3

Our broken behavior:

Batch 1 arrives: producer's baseOffset = 0
Broker stores:   at internal offset 0, bytes unchanged
                 Increments next_offset by 1 (wrong!)

Batch 2 arrives: producer's baseOffset = 0
Broker stores:   at internal offset 1, bytes unchanged (baseOffset still 0!)
                 Increments next_offset by 1

When the consumer reads batch 2, it sees baseOffset = 0, thinks it's getting duplicate data from offset 0, and things go completely sideways.

The fix

Once I understood the problem, the fix became clear. According to the Kafka Protocol Guide, the Produce API response includes a base_offset field—"The base offset" assigned to the batch by the broker. This tells us the broker is responsible for assigning and returning this offset.

We needed a new method that understands Kafka batch semantics:

/// Append a Kafka record batch to this partition
///
/// This method handles Kafka record batches which may contain multiple records.
/// It atomically reserves `record_count` offsets and updates the batch header's
/// baseOffset field (first 8 bytes, big-endian) to the broker-assigned offset.
///
/// This is critical for Kafka protocol compliance - the broker MUST update
/// the baseOffset in the batch header before storing.
pub fn append_batch(&mut self, batch_data: Bytes, record_count: i32) -> Result<i64> {
    if record_count <= 0 {
        return Err(StreamlineError::Storage(
            "Record count must be positive".to_string(),
        ));
    }
 
    // Atomically reserve offsets for ALL records in the batch
    let base_offset = self.next_offset.fetch_add(record_count as i64, Ordering::SeqCst);
    let timestamp = chrono::Utc::now().timestamp_millis();
 
    // Update the Kafka batch header's baseOffset (first 8 bytes, big-endian)
    // According to Kafka protocol, the broker must set this to the actual log offset
    let mut batch_bytes = bytes::BytesMut::from(batch_data.as_ref());
    if batch_bytes.len() >= 8 {
        batch_bytes[0..8].copy_from_slice(&base_offset.to_be_bytes());
    }
    let value = batch_bytes.freeze();
 
    // Store the batch as a single Record
    let record = Record::new(base_offset, timestamp, None, value);
 
    // ... storage logic (segment files or in-memory) ...
 
    Ok(base_offset)
}

The key operations:

1. Atomic offset reservation: fetch_add(record_count as i64) reserves N offsets in one atomic operation. No race conditions.

2. Header modification: Bytes 0-7 get overwritten with the broker-assigned base_offset in big-endian format. This is the critical step we were missing.

3. Return the base_offset so the produce response can include it.

Updated produce handler

// Append the Kafka record batch using append_batch
let batch_data = Bytes::copy_from_slice(records);
let effective_record_count = if record_count > 0 { record_count } else { 1 };
 
match self
    .topic_manager
    .append_batch(&topic_name, partition_index, batch_data, effective_record_count)
{
    Ok(base_offset) => {
        // Update LEO (Log End Offset) correctly
        if let Some(ref replication_manager) = self.replication_manager {
            let new_leo = base_offset + effective_record_count as i64;
            replication_manager
                .set_log_end_offset(&topic_name, partition_index, new_leo)
                .await;
        }
        (base_offset, NONE)
    }
    Err(e) => {
        error!(topic = %topic_name, partition = partition_index, error = %e, 
               "Failed to append batch");
        (-1, OFFSET_OUT_OF_RANGE)
    }
}

Validation

After the fix:

# Send 11 messages
$ echo -e "msg1\nmsg2\nmsg3\nmsg4\nmsg5\nmsg6\nmsg7\nmsg8\nmsg9\nmsg10\nmsg11" | \
    kafka-console-producer --bootstrap-server localhost:9092 --topic test-topic
 
# All 11 received!
$ kafka-console-consumer --bootstrap-server localhost:9092 \
    --topic test-topic --from-beginning
msg1
msg2
msg3
msg4
msg5
msg6
msg7
msg8
msg9
msg10
msg11
Processed a total of 11 messages

And with multiple batches:

# Send batch 1 (3 messages)
$ echo -e "batch1_msg1\nbatch1_msg2\nbatch1_msg3" | kafka-console-producer ...
 
# Send batch 2 (3 messages)
$ echo -e "batch2_msg1\nbatch2_msg2\nbatch2_msg3" | kafka-console-producer ...
 
# All 6 messages, in order
$ kafka-console-consumer --from-beginning ...
batch1_msg1
batch1_msg2
batch1_msg3
batch2_msg1
batch2_msg2
batch2_msg3
Processed a total of 6 messages

Server logs confirmed correct offset tracking:

DEBUG Kafka batch appended topic=test-topic partition=0 base_offset=0 record_count=11
DEBUG Kafka batch appended topic=test-topic partition=0 base_offset=11 record_count=3

Lessons learned

1. Protocol specs are contracts, not suggestions

The Kafka documentation says:

"The broker will update the baseOffset field in each RecordBatch..."

This isn't optional. It's not "you can update it if you want." It's a contract. Clients depend on this behavior. When I violated it, there were no errors—just silent data loss.

For your protocol implementations, this means: Read specs like legal documents. Every "must," "will," and "shall" is a requirement you need to implement.

2. Partial success masks total failure

The first batch worked because:

• Producer sent baseOffset = 0
• We happened to assign offset 0
• We happened to increment to 1 (wrong, but irrelevant for the first batch)

This made it look like things were working. The bug only manifested on the second batch, or when a consumer tried to resume from a previous position.

For your testing strategy, this means: Test multi-batch scenarios. Test resume. Test the second request, not just the first.

3. Your abstractions must match the protocol's model

Our storage layer had a simple model: each append() is one record, one offset. Kafka's model is different: each append is a batch, multiple offsets. This mismatch was the root cause.

We had two choices:

1. Change our storage abstraction to understand batches
2. Translate at the protocol layer

We chose option 1 (append_batch) because it's semantically correct—we really are storing a batch, not a single record.

For your architecture decisions, this means: When implementing protocol compatibility, your abstractions should either match the protocol's model or have an explicit, well-tested translation layer.

4. Trace data end-to-end

The bug wasn't in the network code. It wasn't in the storage code. It wasn't in the fetch code. It was in how they interacted:

Produce:  Store batch at offset 0 (should reserve 0-10)
Storage:  Increment offset by 1 (should be 11)
Fetch:    Return batch with baseOffset=0 (correct)
Consumer: Request offset 11 (correct per protocol)
Storage:  No data at offset >= 11 (BUG!)

For your debugging workflow, this means: When debugging protocol issues, trace a single request through every layer. Draw the sequence diagram. The bug is often at a layer boundary.

5. Byte layouts matter

The fix was literally changing 8 bytes at the start of the batch:

batch_bytes[0..8].copy_from_slice(&base_offset.to_be_bytes());

Eight bytes. Big-endian. At offset 0. That's it.

But to know those 8 bytes needed to change, I had to understand:

• The Record Batch binary format
• That the broker is responsible for writing the baseOffset
• How consumers use baseOffset + offsetDelta
• The relationship between lastOffsetDelta and batch size

For your protocol work, this means: Protocol implementation requires byte-level understanding. Hexdump is your friend. The spec is your bible.

Debugging checklist for protocol implementations

For future me (and you), here's what I'll check next time:

Data flow analysis

• Trace a single request from client to storage and back
• Log the exact bytes at each layer boundary
• Verify data isn't being transformed unexpectedly

Protocol compliance

• Re-read the protocol specification for the failing operation
• Check if you're handling all required fields
• Verify you're updating mutable fields the protocol requires

Offset/sequence handling

• Verify offset calculations match protocol expectations
• Check for off-by-one errors
• Ensure atomic operations where required

Multi-batch testing

• Test with multiple sequential batches
• Test consumer resume from various offsets
• Test producer sending, stopping, sending again

The real takeaway: implementation as education

This bug took hours to find but minutes to fix. The ratio of investigation time to code changed is probably 80:1. That's normal for protocol implementation bugs.

The fix was 15 lines of code. But those 15 lines required understanding:

• Kafka's 61-byte Record Batch header layout
• The broker's contract to update baseOffset
• How consumers use baseOffset + offsetDelta for offset calculation
• Atomic offset reservation for multi-record batches

Here's the thing: I've used Kafka in production for years. I've configured producers, tuned consumer groups, debugged rebalancing storms, and stared at lag metrics at 2 AM. But I never truly understood Kafka until I tried to implement its protocol.

Now when I see a consumer "lose its place" in production, I don't just restart it and hope for the best—I understand exactly what offset calculations are happening and where they might go wrong. When I configure fetch.max.bytes, I know how that interacts with batch sizes at the byte level.

If you want to level up your understanding of any distributed system:

1. Pick a protocol you use regularly (Kafka, Redis, PostgreSQL wire protocol)
2. Try implementing basic compatibility—even just enough to handle a single request type
3. Watch your mental model transform from "magic box" to "bytes on wire"

The bugs you hit will teach you more than any documentation ever could.

If you're implementing Kafka compatibility—or any complex protocol—I hope this helps you avoid the same pitfall. And if you hit a similar "partial success with silent data loss" bug, now you know where to look: at the byte level, at the layer boundaries, and at what the protocol requires you to do (not just what you think you need to do).

Happy debugging!

Sources and References

Kafka Protocol Documentation

• Record Batch Format (Section 5.3.1): Kafka Message Format - The on-disk format of RecordBatch with field-by-field specification
• Wire Protocol: Kafka Protocol Specification - Complete wire protocol documentation including Produce/Fetch API definitions
• Message Format Overview: Kafka Documentation - High-level overview of message and batch structure
• KIP-98: Exactly Once Delivery and Transactional Messaging - The KIP that introduced the current record batch format

Reference Implementation

• Batch Header Layout: DefaultRecordBatch.java - Kafka's reference implementation with byte offset constants

API Details

• Produce API (Key: 0): Returns base_offset field in response—the offset assigned by the broker
• Fetch API (Key: 1): Uses fetch_offset to request data from a specific position