Required for hanzoai/.github/.github/workflows/docker-build.yml@main —
without it the workflow_call dies as startup_failure with no jobs
dispatched. Caller permissions are a CEILING.
Admin endpoints now served under /v1/stream/{status,topics,groups}
following the /<version>/<service>/<path> convention. Root /
redirects to /v1/stream/. Added /healthz for K8s HTTP probes.
- Fix NATS JetStream subject mismatch: SubjectName now returns
StreamName + ".data" to match existing stream subjects, fixing
capture produce failures ("Unknown broker error")
- Fix OffsetFetch flexible version threshold (v6, not v8) and add
version-aware encoding for v0-v5 (non-flexible) vs v6-v7 (flexible)
- Fix ApiVersions to use non-flexible encoding for v0-v2
- Add offset out-of-range detection in Fetch with proper error code
- Add offset validation and corruption purge in consumer offsets
- Add admin HTTP server on :9093 with /status, /topics, /groups
endpoints for Kafka-friendly monitoring without raw NATS access
- Reduce Fetch/Heartbeat/ListOffsets request logging to DEBUG
- Set default log level to INFO
- Log RecordBatch hex on Produce for debugging
rdkafka interprets PreferredReadReplica=0 as "redirect to broker 0"
and skips all message data. Use -1 (no preferred replica) so messages
are delivered to consumers correctly.
Reflection-based Decode() uses compact format, but v1-v2 are non-flexible
(int16 strings) and v3 uses compact but not the CoordinatorKeys array
layout. Decode manually for each version range.
- Fetch: non-flexible decode/encode for v0-v11, flexible for v12+
- FindCoordinator: v0 (no KeyType), v1-v2 flat non-flexible, v3 flat flexible, v4+ array
- ApiVersions: advertise MinVersion=0 for Fetch and FindCoordinator so older clients can connect
The response encoder was always using flexible format (compact arrays, compact
strings, tagged field terminators) for ALL responses. This caused rdkafka to
reject Produce v7 responses with "BadMessage" because v7 expects non-flexible
format (int32 arrays, int16 strings, no tagged fields).
- Add EncodeNonFlexible() and PutNonFlexible() for non-flexible response encoding
- Add PutArrayLen() (int32) and PutNullableString() (int16) encoder helpers
- Make EncodeResponseBytes() auto-detect flexible vs non-flexible based on API version
- Write manual encodeProduceResponse() with per-version field handling (v7 omits
record_errors/error_message, v2+ adds log_append_time, v5+ adds log_start_offset)
- Write manual encodeMetadataResponse() for non-flexible versions (v0-v8)
- Add encoder tests for both flexible and non-flexible response formats
Kafka non-flexible NULLABLE_STRING uses 0xFFFF (-1 as int16) for null.
The existing String() decoder treated 0xFFFF as length 65535, causing
out-of-range panics on Produce v7 transactional_id (which is null).
Added NullableString() to properly handle null (-1) and empty (0) cases.
Produce v7 uses non-compact encoding (int16 strings, int32 arrays/bytes)
while v9+ uses compact encoding. The decoder now branches on version,
matching the same pattern as the Metadata decoder fix.
Per Kafka protocol spec, request header v2 (flexible) uses the same
int16-prefixed NULLABLE_STRING for client_id as header v1. Only the
TAG_BUFFER is added after client_id in v2. Previous commit incorrectly
read client_id as compact string (uvarint) for flexible requests.
Also add manual Metadata request decoder that handles both v4 and v12
body formats, and null client_id test case.
- ParseHeader now detects flexible version requests per API key/version
- Flexible headers: compact client_id (uvarint) + tagged fields skip
- Non-flexible headers: int16 client_id, no tagged fields
- Add manual Metadata request decoder for v4 and v12 formats
- Add panic recovery in HandleConnection to prevent crashes
- Add ParseHeader unit tests for both header formats
rdkafka sends ApiVersions v3+ requests with tagged fields in the
header. The stream service was panicking on non-empty _tagged_fields,
causing crash loops when used with PostHog's capture service.
Now properly skips tagged fields (varint-encoded tag + compact bytes)
rather than crashing.
hanzo-dev