Add Kafka producer protocol support

[carlhoerberg]

Summary

• Implements Kafka protocol support for producers, mapping Kafka topics 1:1 to LavinMQ stream queues
• Follows the existing MQTT implementation pattern
• Adds new src/lavinmq/kafka/ module with protocol handling

Features

• ApiVersions API - Clients discover supported API versions
• Metadata API - Clients query broker info and topic (stream queue) metadata
• Produce API - Clients send messages that are stored in stream queues
• Auto-create topics - Stream queues are auto-created when producing to non-existent topics
• Single partition - Topics have partition 0 only (simplifies implementation)
• Topic = Stream Queue - Direct 1:1 mapping by name

Configuration

[kafka]
bind = 127.0.0.1
port = 9092              # Enabled by default
tls_port = -1
unix_path =
default_vhost = /
auto_create_topics = true

Limitations (by design for initial implementation)

• Producer only (no consumer support yet)
• No authentication (SASL PLAIN can be added later)
• Single partition only
• No compression support in record batches
• No transactional/idempotent producer support

Test plan

• All existing tests pass (1184 examples, 0 failures)
• New protocol tests for ApiVersions, Metadata, and Produce
• Manual testing with Python Kafka client

🤖 Generated with Claude Code

[claude]

Claude finished @carlhoerberg's task —— View job

---

Code Review - Issues Found

I've reviewed the Kafka producer protocol implementation and found several critical issues that need fixing:

Bugs

1. Thread Safety Issue in RecordBatch Parsing - src/lavinmq/kafka/record_batch.cr:12

@@headers_reuse_buffer = Hash(String, Bytes).new

This class variable is shared across all parsing calls and lacks synchronization. The code even has a comment acknowledging it's "not thread-safe for concurrent parsing" but doesn't address the issue. In a high-performance message broker, this could cause data corruption when multiple connections parse records simultaneously.

2. Missing CRC Validation - src/lavinmq/kafka/record_batch.cr:134,239
The code reads the CRC field but never validates it against the batch data. This is a serious protocol compliance issue that could allow corrupted data to be processed silently:

crc = io.read_bytes(UInt32, ::IO::ByteFormat::BigEndian)
# CRC is read but never validated

3. Request Size Validation Logic Error - src/lavinmq/kafka/protocol.cr:25-27

if slice.size > @bytes_remaining
  raise MaxRequestSizeError.new("Attempt to read beyond request size")
end

This check occurs in the read() method but doesn't handle partial reads correctly. If io.read() returns fewer bytes than requested, @bytes_remaining becomes incorrectly negative.

4. Resource Leak in Error Conditions - src/lavinmq/kafka/record_batch.cr:277-301
Decompression creates IO objects that may not be properly closed in error scenarios:

decompressed = ::IO::Memory.new
Compress::Gzip::Reader.open(::IO::Memory.new(data)) do |gzip|
  ::IO.copy(gzip, decompressed)
end

Performance Issues

5. Inefficient String Conversion - src/lavinmq/kafka/client.cr:385,393

@kafka_headers_hash[key] = String.new(value)  # Creates string copy
@kafka_headers_hash["kafka-key"] = String.new(key) # Another copy

These String.new(Bytes) calls create unnecessary memory copies in the hot path for every message.

6. Reused Hash Not Pre-sized - src/lavinmq/kafka/record_batch.cr:12
The reused headers buffer should have initial_capacity set to avoid rehashing for typical header counts.

Security Issues

7. DoS via Compression Bomb - src/lavinmq/kafka/record_batch.cr:277-301
No limits on decompressed data size. Malicious clients could send small compressed payloads that expand to gigabytes, consuming all server memory.

8. Memory Exhaustion via Large Arrays - src/lavinmq/kafka/protocol.cr:142-144

Array.new(length) { yield }

No bounds checking on array lengths read from network. Malicious clients could request arrays with length = 2**31 - 1.

Crystal Anti-patterns

9. Class Variable for Instance State - src/lavinmq/kafka/record_batch.cr:12
Using @@ for what should be instance or fiber-local state violates Crystal conventions and creates the thread safety issue.

10. Missing Error Types - src/lavinmq/kafka/record_batch.cr:297
Generic Error.new() instead of specific exception types makes error handling difficult for callers.

---