Version-aware response encoding for non-flexible Kafka protocol versions
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
This commit is contained in:
Hanzo Dev
+77
-2
@@ -151,6 +151,81 @@ func (b *Broker) getMetadataResponse(req types.Request) []byte {
|
||||
Topics: topics,
|
||||
}
|
||||
log.Debug("MetadataResponse %+v", response)
|
||||
encoder := serde.NewEncoder()
|
||||
return encoder.EncodeResponseBytes(req, response)
|
||||
return encodeMetadataResponse(req, response)
|
||||
}
|
||||
|
||||
// encodeMetadataResponse manually encodes the Metadata response based on API version.
|
||||
// v0-v8: non-flexible (int32 arrays, int16 strings, no tagged fields).
|
||||
// v9+: flexible (compact arrays/strings, tagged field terminators).
|
||||
func encodeMetadataResponse(req types.Request, response MetadataResponse) []byte {
|
||||
e := serde.NewEncoder()
|
||||
e.PutInt32(req.CorrelationID)
|
||||
|
||||
if req.RequestAPIVersion >= 9 {
|
||||
// Flexible response
|
||||
e.EndStruct()
|
||||
e.Encode(response)
|
||||
} else {
|
||||
// Non-flexible response
|
||||
if req.RequestAPIVersion >= 3 {
|
||||
e.PutInt32(response.ThrottleTimeMs)
|
||||
}
|
||||
// brokers array
|
||||
e.PutArrayLen(len(response.Brokers))
|
||||
for _, broker := range response.Brokers {
|
||||
e.PutInt32(broker.NodeID)
|
||||
e.PutString(broker.Host)
|
||||
e.PutInt32(broker.Port)
|
||||
if req.RequestAPIVersion >= 1 {
|
||||
e.PutNullableString(broker.Rack)
|
||||
}
|
||||
}
|
||||
if req.RequestAPIVersion >= 2 {
|
||||
e.PutNullableString(response.ClusterID)
|
||||
}
|
||||
if req.RequestAPIVersion >= 1 {
|
||||
e.PutInt32(response.ControllerID)
|
||||
}
|
||||
// topics array
|
||||
e.PutArrayLen(len(response.Topics))
|
||||
for _, topic := range response.Topics {
|
||||
e.PutInt16(topic.ErrorCode)
|
||||
e.PutString(topic.Name)
|
||||
if req.RequestAPIVersion >= 1 {
|
||||
e.PutBool(topic.IsInternal)
|
||||
}
|
||||
// partitions array
|
||||
e.PutArrayLen(len(topic.Partitions))
|
||||
for _, p := range topic.Partitions {
|
||||
e.PutInt16(p.ErrorCode)
|
||||
e.PutInt32(p.PartitionIndex)
|
||||
e.PutInt32(p.LeaderID)
|
||||
if req.RequestAPIVersion >= 7 {
|
||||
e.PutInt32(p.LeaderEpoch)
|
||||
}
|
||||
// replica nodes
|
||||
e.PutArrayLen(len(p.ReplicaNodes))
|
||||
for _, r := range p.ReplicaNodes {
|
||||
e.PutInt32(r)
|
||||
}
|
||||
// isr nodes
|
||||
e.PutArrayLen(len(p.IsrNodes))
|
||||
for _, r := range p.IsrNodes {
|
||||
e.PutInt32(r)
|
||||
}
|
||||
if req.RequestAPIVersion >= 5 {
|
||||
// offline replicas
|
||||
e.PutArrayLen(len(p.OfflineReplicas))
|
||||
for _, r := range p.OfflineReplicas {
|
||||
e.PutInt32(r)
|
||||
}
|
||||
}
|
||||
}
|
||||
if req.RequestAPIVersion >= 8 {
|
||||
e.PutInt32(topic.TopicAuthorizedOperations)
|
||||
}
|
||||
}
|
||||
}
|
||||
e.PutLen()
|
||||
return e.Bytes()
|
||||
}
|
||||
|
||||
+51
-2
@@ -135,6 +135,55 @@ func (b *Broker) getProduceResponse(req types.Request) []byte {
|
||||
}
|
||||
response.ProduceTopicResponses = append(response.ProduceTopicResponses, produceTopicResponse)
|
||||
}
|
||||
encoder := serde.NewEncoder()
|
||||
return encoder.EncodeResponseBytes(req, response)
|
||||
return encodeProduceResponse(req, response)
|
||||
}
|
||||
|
||||
// encodeProduceResponse manually encodes the Produce response based on API version.
|
||||
// v0-v8: non-flexible format (int32 arrays, int16 strings, no tagged fields).
|
||||
//
|
||||
// v7 fields: index, error_code, base_offset, log_append_time_ms, log_start_offset
|
||||
// v8 adds: record_errors, error_message
|
||||
//
|
||||
// v9+: flexible format (compact arrays/strings, tagged field terminators).
|
||||
func encodeProduceResponse(req types.Request, response ProduceResponse) []byte {
|
||||
e := serde.NewEncoder()
|
||||
e.PutInt32(req.CorrelationID)
|
||||
|
||||
if req.RequestAPIVersion >= 9 {
|
||||
// Flexible response header: tagged fields after correlation_id
|
||||
e.EndStruct()
|
||||
e.Encode(response)
|
||||
} else {
|
||||
// Non-flexible response: manual encoding without tagged fields
|
||||
// topics array (int32 count)
|
||||
e.PutArrayLen(len(response.ProduceTopicResponses))
|
||||
for _, topic := range response.ProduceTopicResponses {
|
||||
e.PutString(topic.Name)
|
||||
// partitions array (int32 count)
|
||||
e.PutArrayLen(len(topic.ProducePartitionResponses))
|
||||
for _, p := range topic.ProducePartitionResponses {
|
||||
e.PutInt32(p.Index)
|
||||
e.PutInt16(p.ErrorCode)
|
||||
e.PutInt64(p.BaseOffset)
|
||||
if req.RequestAPIVersion >= 2 {
|
||||
e.PutInt64(p.LogAppendTimeMs)
|
||||
}
|
||||
if req.RequestAPIVersion >= 5 {
|
||||
e.PutInt64(p.LogStartOffset)
|
||||
}
|
||||
if req.RequestAPIVersion >= 8 {
|
||||
// record_errors: empty array
|
||||
e.PutArrayLen(len(p.RecordErrors))
|
||||
// error_message: nullable string
|
||||
e.PutNullableString(p.ErrorMessage)
|
||||
}
|
||||
}
|
||||
}
|
||||
if req.RequestAPIVersion >= 1 {
|
||||
e.PutInt32(response.ThrottleTimeMs)
|
||||
}
|
||||
}
|
||||
|
||||
e.PutLen()
|
||||
return e.Bytes()
|
||||
}
|
||||
|
||||
+86
-3
@@ -200,20 +200,103 @@ func (e *Encoder) EndStruct() {
|
||||
e.offset++
|
||||
}
|
||||
|
||||
// PutArrayLen encodes a non-flexible array length as int32
|
||||
func (e *Encoder) PutArrayLen(l int) {
|
||||
e.PutInt32(uint32(l))
|
||||
}
|
||||
|
||||
// PutNullableString encodes a NULLABLE_STRING (int16 length prefix, -1 = null)
|
||||
func (e *Encoder) PutNullableString(s string) {
|
||||
if s == "" {
|
||||
e.PutInt16(0xFFFF) // null
|
||||
return
|
||||
}
|
||||
e.ensureBufferSpace(2 + len(s))
|
||||
e.PutInt16(uint16(len(s)))
|
||||
copy(e.b[e.offset:], s)
|
||||
e.offset += len(s)
|
||||
}
|
||||
|
||||
// Bytes returns the encoded data as a byte slice
|
||||
func (e *Encoder) Bytes() []byte {
|
||||
return e.b[:e.offset]
|
||||
}
|
||||
|
||||
// EncodeResponseBytes serializes encodes the response into bytes
|
||||
// IsFlexibleResponse returns true if the given API key + version uses flexible encoding for responses.
|
||||
func IsFlexibleResponse(apiKey, apiVersion uint16) bool {
|
||||
return isFlexibleRequest(apiKey, apiVersion)
|
||||
}
|
||||
|
||||
// EncodeResponseBytes serializes encodes the response into bytes.
|
||||
// For flexible versions, includes tagged field terminators.
|
||||
// For non-flexible versions, uses plain encoding without tagged fields.
|
||||
func (e *Encoder) EncodeResponseBytes(req types.Request, response any) []byte {
|
||||
e.PutInt32(req.CorrelationID)
|
||||
e.EndStruct()
|
||||
e.Encode(response)
|
||||
if IsFlexibleResponse(req.RequestAPIKey, req.RequestAPIVersion) {
|
||||
e.EndStruct() // response header tagged fields
|
||||
e.Encode(response)
|
||||
} else {
|
||||
e.EncodeNonFlexible(response)
|
||||
}
|
||||
e.PutLen()
|
||||
return e.Bytes()
|
||||
}
|
||||
|
||||
// EncodeNonFlexible encodes a struct using non-flexible Kafka format:
|
||||
// int32 array counts, int16 strings, no tagged field terminators.
|
||||
func (e *Encoder) EncodeNonFlexible(x any) {
|
||||
t := reflect.TypeOf(x)
|
||||
v := reflect.ValueOf(x)
|
||||
|
||||
for i := 0; i < t.NumField(); i++ {
|
||||
field := t.Field(i)
|
||||
value := v.Field(i)
|
||||
if field.Type.Kind() == reflect.Slice {
|
||||
e.PutArrayLen(value.Len()) // int32 count
|
||||
if field.Type.Elem().Kind() == reflect.Struct {
|
||||
for j := 0; j < value.Len(); j++ {
|
||||
e.EncodeNonFlexible(value.Index(j).Interface())
|
||||
}
|
||||
} else {
|
||||
for j := 0; j < value.Len(); j++ {
|
||||
e.PutNonFlexible(value.Index(j).Interface())
|
||||
}
|
||||
}
|
||||
} else if field.Type.Kind() == reflect.Struct {
|
||||
e.EncodeNonFlexible(value.Interface())
|
||||
} else {
|
||||
e.PutNonFlexible(value.Interface())
|
||||
}
|
||||
}
|
||||
// No EndStruct — non-flexible has no tagged fields
|
||||
}
|
||||
|
||||
// PutNonFlexible encodes a value using non-flexible format (int16 strings instead of compact)
|
||||
func (e *Encoder) PutNonFlexible(i any) {
|
||||
switch c := i.(type) {
|
||||
case bool:
|
||||
e.PutBool(c)
|
||||
case uint8:
|
||||
e.PutInt8(c)
|
||||
case uint16:
|
||||
e.PutInt16(c)
|
||||
case uint32:
|
||||
e.PutInt32(c)
|
||||
case uint64:
|
||||
e.PutInt64(c)
|
||||
case string:
|
||||
e.PutString(c) // int16 length, not compact
|
||||
case []byte:
|
||||
// int32 length prefix + bytes
|
||||
e.PutInt32(uint32(len(c)))
|
||||
e.PutBytes(c)
|
||||
case [16]byte:
|
||||
e.PutBytes(c[:])
|
||||
default:
|
||||
log.Panic("Unknown type %T", c)
|
||||
}
|
||||
}
|
||||
|
||||
// FinishAndReturn finishes the encoding and returns the final byte slice
|
||||
func (e *Encoder) FinishAndReturn() []byte {
|
||||
e.EndStruct() // close struct
|
||||
|
||||
@@ -3,6 +3,8 @@ package serde
|
||||
import (
|
||||
"encoding/binary"
|
||||
"testing"
|
||||
|
||||
"github.com/hanzoai/stream/types"
|
||||
)
|
||||
|
||||
func TestParseHeader_NonFlexible(t *testing.T) {
|
||||
@@ -152,6 +154,78 @@ func TestParseHeader_ApiVersionsAlwaysNonFlexible(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestEncodeResponseBytes_NonFlexible(t *testing.T) {
|
||||
// Produce v7 (non-flexible) response should NOT have tagged field terminators
|
||||
type SimpleResponse struct {
|
||||
ErrorCode uint16
|
||||
Name string
|
||||
}
|
||||
req := types.Request{
|
||||
RequestAPIKey: 0, // Produce
|
||||
RequestAPIVersion: 7, // non-flexible (threshold is v9)
|
||||
CorrelationID: 42,
|
||||
}
|
||||
resp := SimpleResponse{ErrorCode: 0, Name: "test-topic"}
|
||||
encoder := NewEncoder()
|
||||
result := encoder.EncodeResponseBytes(req, resp)
|
||||
|
||||
// Parse the result: int32 length prefix + int32 correlation_id + body
|
||||
if len(result) < 8 {
|
||||
t.Fatalf("result too short: %d bytes", len(result))
|
||||
}
|
||||
totalLen := Encoding.Uint32(result[0:4])
|
||||
if int(totalLen) != len(result)-4 {
|
||||
t.Fatalf("length prefix mismatch: got %d, expected %d", totalLen, len(result)-4)
|
||||
}
|
||||
corrID := Encoding.Uint32(result[4:8])
|
||||
if corrID != 42 {
|
||||
t.Fatalf("correlation_id: expected 42, got %d", corrID)
|
||||
}
|
||||
// Next should be ErrorCode (uint16 = 0), NOT a 0x00 tagged field terminator
|
||||
// followed by Name as int16-prefixed string (NOT compact string)
|
||||
off := 8
|
||||
errCode := Encoding.Uint16(result[off:])
|
||||
off += 2
|
||||
if errCode != 0 {
|
||||
t.Fatalf("error_code: expected 0, got %d", errCode)
|
||||
}
|
||||
nameLen := Encoding.Uint16(result[off:])
|
||||
off += 2
|
||||
if nameLen != 10 { // "test-topic" = 10 chars
|
||||
t.Fatalf("name length: expected 10 (int16 format), got %d", nameLen)
|
||||
}
|
||||
name := string(result[off : off+int(nameLen)])
|
||||
if name != "test-topic" {
|
||||
t.Fatalf("name: expected 'test-topic', got %q", name)
|
||||
}
|
||||
}
|
||||
|
||||
func TestEncodeResponseBytes_Flexible(t *testing.T) {
|
||||
// Produce v9 (flexible) response SHOULD have tagged field terminator after correlation_id
|
||||
type SimpleResponse struct {
|
||||
ErrorCode uint16
|
||||
}
|
||||
req := types.Request{
|
||||
RequestAPIKey: 0, // Produce
|
||||
RequestAPIVersion: 9, // flexible
|
||||
CorrelationID: 99,
|
||||
}
|
||||
resp := SimpleResponse{ErrorCode: 0}
|
||||
encoder := NewEncoder()
|
||||
result := encoder.EncodeResponseBytes(req, resp)
|
||||
|
||||
off := 4 // skip length prefix
|
||||
corrID := Encoding.Uint32(result[off:])
|
||||
off += 4
|
||||
if corrID != 99 {
|
||||
t.Fatalf("correlation_id: expected 99, got %d", corrID)
|
||||
}
|
||||
// Next byte should be 0x00 (tagged fields terminator for response header)
|
||||
if result[off] != 0 {
|
||||
t.Fatalf("expected tagged field terminator 0x00 at offset %d, got 0x%02X", off, result[off])
|
||||
}
|
||||
}
|
||||
|
||||
func TestParseHeader_NullClientID(t *testing.T) {
|
||||
// Flexible request with null client_id (int16 = -1 = 0xFFFF)
|
||||
bodyPayload := []byte{0x42}
|
||||
|
||||
Reference in New Issue
Block a user