Stream topic and partition data instead of buffering by andreterroir · Pull Request #13 · andreterroir/klog

andreterroir · 2026-06-02T09:32:57Z

Refactor the Kafka protocol implementation to stream topic and partition data one entry at a time, eliminating the need to buffer entire arrays in memory.

Summary

This change restructures how ProduceRequest topic and partition data are read and written. Instead of treating them as complete arrays that must be buffered, they are now streamed entry-by-entry using the size prefixes already present in the protocol.

Key changes

New protocol types: Introduce Topic and Partition structs that represent the fixed-size header of each topic_data and partition_data entry, with the caller responsible for consuming variable-length records separately
Streaming reader functions: reader.topic_data() and reader.partition_data() now read and return a single entry's fixed part, allowing callers to process entries sequentially without buffering
Streaming writer functions: writer.topic_data() and writer.partition_data() now write a single entry instead of entire arrays
Server implementation: klog.zig now iterates through topics and partitions using the counts from each level, consuming records in fixed-size chunks via a new log_records() helper
Client implementation: client.zig updated to write topic and partition headers followed by their data, mirroring the streaming pattern
Test updates: Simplified tests to verify round-trip behavior of individual entries rather than full arrays

Implementation details

The protocol's compact size encoding already provides the necessary structure: topic_data_size tells how many topics follow, each topic's partition_count tells how many partitions follow. By reading these counts upfront, the server can stream through the request without buffering, consuming records in configurable chunks (4096 bytes in the example).

This approach scales to arbitrarily large record payloads while keeping memory usage bounded.

https://claude.ai/code/session_01Sgm6nVAriGJDHANn7yLTnK

Add streaming readers for the variable-length topic_data array of a Produce request. reader.topic_data reads a topic_id UUID plus its partition count, and reader.partition_data reads a partition index plus the records byte length. The record bytes are left in the stream for the caller, so an arbitrarily large request is processed one topic/partition at a time instead of being buffered whole. The server's produce() now loops over topic_data_size topics and their partitions, logging each topic/partition and draining the records in fixed-size chunks (logging the byte count and a short hex preview). To mirror the streaming readers, the writer's topic_data/partition_data are reworked from whole-array helpers into per-element ones: the topic array length stays in produce_req, the partition count moves into the topic header. Their byte-exact tests become round-trip tests against the new readers, matching the file's prevailing style. The client now writes example data for two topics with one and two partitions. https://claude.ai/code/session_01Sgm6nVAriGJDHANn7yLTnK

Make writer.topic_data take a Topic struct so both writer helpers now take structs (partition_data already took PartitionData), mirroring the reader side which returns Topic/Partition. The client organizes its example data in TopicData/PartitionData structs and iterates, deriving topic_data_size from the array length instead of hardcoding it. No change to the wire format. https://claude.ai/code/session_01Sgm6nVAriGJDHANn7yLTnK

Replace the {01 02 03} partition records with feedface so all three example payloads (cafebabe, deadbeef, feedface) read as recognizable hex words. https://claude.ai/code/session_01Sgm6nVAriGJDHANn7yLTnK

Delete the writer-only TopicData/PartitionData structs and use the reader's Topic/Partition for both directions. PartitionData embedded the variable record bytes, which forced a separate size-only struct for the streaming reader; moving those bytes out of the struct lets a single Partition { index, records_size } serve both. writer.partition_data now writes only the index and the COMPACT_NULLABLE_RECORDS length prefix; the caller writes the record bytes itself, mirroring how reader.partition_data returns the size and leaves the bytes in the stream. The client gains a small write_partition helper that derives the size from the slice. Wire format is unchanged. https://claude.ai/code/session_01Sgm6nVAriGJDHANn7yLTnK

Add a third partition to topic two with null records so the client also drives the COMPACT_NULLABLE_RECORDS null branch end to end (the server logs it as "null records"). https://claude.ai/code/session_01Sgm6nVAriGJDHANn7yLTnK

Rename Topic/Partition to TopicData/PartitionData so the struct names match the protocol's topic_data/partition_data fields. Pure rename, no behavior change. https://claude.ai/code/session_01Sgm6nVAriGJDHANn7yLTnK

@as

expectEqual resolves expected/actual to a common type via peer type resolution, so bare literals (and records.len / null) suffice in the asserts, matching the existing expectEqual(3, read.topic_data_size) style. The remaining @as in unsigned_varint are implementation widening, not asserts. https://claude.ai/code/session_01Sgm6nVAriGJDHANn7yLTnK

andreterroir · 2026-06-02T13:47:56Z

+        const len = try unsigned_varint(r);
+        return .{
+            .index = index,
+            .records_size = if (len == 0) null else len - 1,


Use the helper function.

Done in dfb6050. Made the reader's compact_size return ?u64 (mirroring the writer's compact_size(?u64)) and call it from partition_data instead of decoding the length inline; also reused it in compact_nullable_str. The non-nullable COMPACT array readers (produce_req topic count, topic_data partition count) reject the 0/null marker as a ProtocolError, which also fixes the previous underflow there.

Generated by Claude Code

The reader's compact_size returned u64 and underflowed on the 0/null marker, so partition_data decoded COMPACT_NULLABLE_RECORDS inline. Make compact_size return ?u64 to mirror the writer's compact_size(?u64), and call it from partition_data (records are nullable) and compact_nullable_str (dropping its duplicated null decode). The non-nullable COMPACT array readers (produce_req topic count, topic_data partition count) reject the null marker as a ProtocolError, which also fixes the previous underflow on a 0 length there. Adds a compact_size null round-trip test. https://claude.ai/code/session_01Sgm6nVAriGJDHANn7yLTnK

andreterroir · 2026-06-02T14:14:34Z

+    try writer.topic_data(io_writer, .{ .topic_id = topic2_uuid, .partition_count = 3 });
+    try write_partition(io_writer, 0, &[_]u8{ 0xde, 0xad, 0xbe, 0xef });
+    try write_partition(io_writer, 1, &[_]u8{ 0xfe, 0xed, 0xfa, 0xce });
+    try write_partition(io_writer, 2, null);


Make the null partition first, to ensure it doesn't break boundaries.

Done in 2c51d3d. Topic two's null-records partition is now written first (index 0), so a partition with records follows it across the records boundary.

Generated by Claude Code

andreterroir · 2026-06-02T14:17:02Z

+fn write_partition(w: *std.Io.Writer, index: i32, records: ?[]const u8) !void {
+    try proto.writer.partition_data(w, .{
+        .index = index,
+        .records_size = if (records) |r| r.len else null,


expose the size helper function from the protocol to be usable here and change its signature to accept an array instead of a number.

Done in e1e91ba. writer.compact_size is now pub and generic over the element type — compact_size(comptime T, ?[]const T) — writing the slice's length as the COMPACT_NULLABLE prefix (or the 0 null marker) instead of taking a count. reader.compact_size is exposed alongside it. To make this work the records length prefix moved out of partition_data: PartitionData is now just { index }, and the client's write_partition writes partition_data (index) → compact_size(u8, records) → the bytes. The non-materialized array counts (topic_data partition_count) write their length via unsigned_varint(+1), matching produce_req.

Generated by Claude Code

Move topic two's null-records partition ahead of the non-null ones so a partition with records follows it across the records boundary, exercising that null (zero-length) records don't desync parsing of what comes next. https://claude.ai/code/session_01Sgm6nVAriGJDHANn7yLTnK

Per review: expose compact_size and have it take the slice (generic over the element type T) rather than a precomputed count, so the client can use it to write a partition's records length. writer.compact_size(comptime T, ?[]const T) writes the slice's length as the COMPACT_NULLABLE prefix (or the 0 null marker); reader.compact_size is exposed alongside it. The records length prefix moves out of partition_data into compact_size: PartitionData is now just { index }, write/read partition_data handle only the index, and the records prefix + bytes are written/consumed separately (client write_partition, server loop). Topic and partition array counts, which aren't materialized as slices, write their length via unsigned_varint(+1) like produce_req already does. https://claude.ai/code/session_01Sgm6nVAriGJDHANn7yLTnK

Drop the comptime element-type parameter; the writer's compact_size takes ?[]const u8 again.

Follow the ProduceRequest pattern uniformly: each struct holds its trailing array's size and writes it when the struct is written, with the elements/bytes streamed after. - TopicData.partition_count -> partition_data_size (size suffix). - PartitionData regains records_size (?u64); partition_data writes the index and the COMPACT_NULLABLE records length prefix, and the caller writes the record bytes after. The writer's compact_size is now used only for compact strings.

The writer now mirrors the reader: compact_size deals in ?u64 (writing N+1, or 0 for the null marker), the exact inverse of reader.compact_size. compact_size_of is a thin wrapper that derives the length from a slice and defers to it. This removes the duplicated +1/null-marker encoding: produce_req, topic_data, and partition_data all write their size fields through the ?u64 primitive (non-nullable counts coerce to the optional), while compact strings use the slice wrapper.

claude added 7 commits June 2, 2026 09:15

Use a readable hex-word for the last example payload

aa61824

Replace the {01 02 03} partition records with feedface so all three example payloads (cafebabe, deadbeef, feedface) read as recognizable hex words. https://claude.ai/code/session_01Sgm6nVAriGJDHANn7yLTnK

Exercise the null-records path in the example client

a89a95a

Add a third partition to topic two with null records so the client also drives the COMPACT_NULLABLE_RECORDS null branch end to end (the server logs it as "null records"). https://claude.ai/code/session_01Sgm6nVAriGJDHANn7yLTnK

Name the shared structs TopicData/PartitionData

b5035cd

Rename Topic/Partition to TopicData/PartitionData so the struct names match the protocol's topic_data/partition_data fields. Pure rename, no behavior change. https://claude.ai/code/session_01Sgm6nVAriGJDHANn7yLTnK

andreterroir commented Jun 2, 2026

View reviewed changes

claude added 5 commits June 2, 2026 14:26

Revert compact_size to a byte-slice signature

77dce96

Drop the comptime element-type parameter; the writer's compact_size takes ?[]const u8 again.

andreterroir merged commit 2971b75 into main Jun 2, 2026
1 check passed

andreterroir deleted the claude/variable-size-produce-reader-3EMfq branch June 2, 2026 20:45

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Stream topic and partition data instead of buffering#13

Stream topic and partition data instead of buffering#13
andreterroir merged 13 commits into
mainfrom
claude/variable-size-produce-reader-3EMfq

andreterroir commented Jun 2, 2026

Uh oh!

andreterroir Jun 2, 2026

Uh oh!

andreterroir Jun 2, 2026

Uh oh!

andreterroir Jun 2, 2026

Uh oh!

andreterroir Jun 2, 2026

Uh oh!

andreterroir Jun 2, 2026

Uh oh!

andreterroir Jun 2, 2026

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Conversation

andreterroir commented Jun 2, 2026

Summary

Key changes

Implementation details

Uh oh!

andreterroir Jun 2, 2026

Choose a reason for hiding this comment

Uh oh!

andreterroir Jun 2, 2026

Choose a reason for hiding this comment

Uh oh!

andreterroir Jun 2, 2026

Choose a reason for hiding this comment

Uh oh!

andreterroir Jun 2, 2026

Choose a reason for hiding this comment

Uh oh!

andreterroir Jun 2, 2026

Choose a reason for hiding this comment

Uh oh!

andreterroir Jun 2, 2026

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants