Stream topic and partition data instead of buffering

client.zig

@@ -25,10 +25,36 @@ pub fn main(init: std.process.Init) !void {
         .client_id = "test-client",
     };
     try writer.req_header(io_writer, req_header);
+
     const req = proto.ProduceRequest{
         .acks = -1, // ISR
         .timeout_ms = 1000,
-        .topic_data_size = 1,
+        .topic_data_size = 2,
     };
     try writer.produce_req(io_writer, req);
+
+    // Example data: topic one with a single partition, topic two with three
+    // (the first with null records, so a non-null partition follows it across
+    // the records boundary). The record bytes live outside the protocol
+    // structs, so each partition is a header write followed by its records.
+    const topic1_uuid = [_]u8{0x11} ** 16;
+    const topic2_uuid = [_]u8{0x22} ** 16;
+
+    try writer.topic_data(io_writer, .{ .topic_id = topic1_uuid, .partition_data_size = 1 });
+    try write_partition(io_writer, 0, &[_]u8{ 0xca, 0xfe, 0xba, 0xbe });
+
+    try writer.topic_data(io_writer, .{ .topic_id = topic2_uuid, .partition_data_size = 3 });
+    try write_partition(io_writer, 0, null);
+    try write_partition(io_writer, 1, &[_]u8{ 0xde, 0xad, 0xbe, 0xef });
+    try write_partition(io_writer, 2, &[_]u8{ 0xfe, 0xed, 0xfa, 0xce });
+}
+
+/// Writes a partition_data entry: the index and records length prefix (both
+/// written by partition_data, the size taken from the slice), then the record
+/// bytes. The bytes live outside the struct, so the size and the bytes are
+/// written from the same slice and always agree.
+fn write_partition(w: *std.Io.Writer, index: i32, records: ?[]const u8) !void {
+    const records_size: ?u64 = if (records) |r| @intCast(r.len) else null;
+    try proto.writer.partition_data(w, .{ .index = index, .records_size = records_size });
+    if (records) |r| try w.writeAll(r);
 }

klog.zig

@@ -58,4 +58,45 @@ fn produce(io: Io, io_reader: *Io.Reader, stream: Io.net.Stream) !void {
     var buf: [max_transactional_id]u8 = undefined;
     const req = try reader.produce_req(io_reader, &buf);
     log.debug("produce request: {}", .{req});
+
+    // Read the topic_data array one entry at a time, using the counts and
+    // sizes that prefix each level rather than buffering the whole request.
+    for (0..req.topic_data_size) |_| {
+        const topic = try reader.topic_data(io_reader);
+        log.info("topic {x}: {d} partition(s)", .{ topic.topic_id, topic.partition_data_size });
+        for (0..topic.partition_data_size) |_| {
+            const partition = try reader.partition_data(io_reader);
+            try log_records(io_reader, partition.index, partition.records_size);
+        }
+    }
+}
+
+/// Logs a partition's records, consuming them from the stream in
+/// fixed-size chunks so an arbitrarily large records blob never has to be
+/// held in memory at once. Logs the byte count and a short hex preview of
+/// the first chunk.
+fn log_records(io_reader: *Io.Reader, index: i32, records_size: ?u64) !void {
+    const size = records_size orelse {
+        log.info("  partition {d}: null records", .{index});
+        return;
+    };
+
+    var chunk: [4096]u8 = undefined;
+    const first_len: usize = @intCast(@min(size, chunk.len));
+    try io_reader.readSliceAll(chunk[0..first_len]);
+    const preview = chunk[0..@min(first_len, 16)];
+    log.info("  partition {d}: {d} record byte(s), first {d}: {x}", .{
+        index,
+        size,
+        preview.len,
+        preview,
+    });
+
+    // Drain the remaining bytes, again one chunk at a time.
+    var remaining = size - first_len;
+    while (remaining > 0) {
+        const take: usize = @intCast(@min(remaining, chunk.len));
+        try io_reader.readSliceAll(chunk[0..take]);
+        remaining -= take;
+    }
 }