mirror of
https://github.com/hanzoai/kafka.git
synced 2026-10-30 14:48:25 +00:00
zeekayandHanzo Dev
55bb7bb42d
test/e2e now spins hanzoai/pubsub/embed plus the broker in-process and drives them with franz-go — no external kafka distribution, no external NATS, one 'go test'. The old harness needed ~/kafka_2.13-3.9.0 binaries and a hand-run server, so it proved nothing anywhere but one laptop. Every 2026-07 outage regression is pinned: compression round-trips, consume across sequence holes, poison messages skipped end-to-end, an e18 committed offset accepted and read back, OFFSET_OUT_OF_RANGE engaging the client's reset policy, group resume across a broker restart, a 1.5MiB record, and two gateways over one store. What the client's strictness flushed out, fixed here: - OffsetFetch with a null topics array means 'everything the group has committed' — now enumerated from the KV bucket (both wire shapes). - LeaveGroup (key 13) is handled and advertised; every client sends it on close and the unknown-key path panicked the connection goroutine. - Serve with port 0 publishes the kernel-assigned port into Metadata/ FindCoordinator and exposes Addr(); Shutdown is idempotent. Co-authored-by: Hanzo Dev <[email protected]>
79 lines
2.7 KiB
Go
79 lines
2.7 KiB
Go
package protocol
|
|
|
|
import (
|
|
log "github.com/hanzoai/kafka/logging"
|
|
"github.com/hanzoai/kafka/serde"
|
|
"github.com/hanzoai/kafka/types"
|
|
)
|
|
|
|
// APIKey represents an API key and its supported version range.
|
|
type APIKey struct {
|
|
APIKey uint16
|
|
MinVersion uint16
|
|
MaxVersion uint16
|
|
}
|
|
|
|
// APIVersionsResponse represents the response for API versions request.
|
|
type APIVersionsResponse struct {
|
|
ErrorCode uint16
|
|
APIKeys []APIKey
|
|
ThrottleTime uint32
|
|
}
|
|
|
|
// APIVersion (Api key = 18)
|
|
// v0-v2: non-flexible response (int32 array counts, no tagged fields).
|
|
// v3+: flexible response (compact arrays, tagged field terminators).
|
|
// Note: ApiVersions request header is always non-flexible (even v3+), but the
|
|
// response switches to flexible format at v3.
|
|
func (b *Broker) getAPIVersionResponse(req types.Request) []byte {
|
|
apiKeys := []APIKey{
|
|
{APIKey: produceKey, MinVersion: 0, MaxVersion: 11},
|
|
{APIKey: fetchKey, MinVersion: 0, MaxVersion: 12},
|
|
{APIKey: listOffsetsKey, MinVersion: 0, MaxVersion: 9},
|
|
{APIKey: metadataKey, MinVersion: 0, MaxVersion: 12},
|
|
{APIKey: offsetCommitKey, MinVersion: 0, MaxVersion: 9},
|
|
{APIKey: offsetFetchKey, MinVersion: 0, MaxVersion: 9},
|
|
{APIKey: findCoordinatorKey, MinVersion: 0, MaxVersion: 6},
|
|
{APIKey: joinGroupKey, MinVersion: 0, MaxVersion: 9},
|
|
{APIKey: heartbeatKey, MinVersion: 0, MaxVersion: 4},
|
|
{APIKey: leaveGroupKey, MinVersion: 0, MaxVersion: 5},
|
|
{APIKey: syncGroupKey, MinVersion: 0, MaxVersion: 5},
|
|
{APIKey: apiVersionKey, MinVersion: 0, MaxVersion: 4},
|
|
{APIKey: createTopicKey, MinVersion: 0, MaxVersion: 7},
|
|
{APIKey: initProducerIDKey, MinVersion: 0, MaxVersion: 5},
|
|
{APIKey: describeConfigsKey, MinVersion: 0, MaxVersion: 4},
|
|
}
|
|
|
|
e := serde.NewEncoder()
|
|
e.PutInt32(req.CorrelationID)
|
|
|
|
if req.RequestAPIVersion >= 3 {
|
|
// ApiVersions response header is ALWAYS v0 (just correlation_id, NO TAG_BUFFER),
|
|
// even for v3+. This is a Kafka protocol special case — the client doesn't know
|
|
// whether flexible headers are supported until after parsing this response.
|
|
// The body itself uses flexible format (compact arrays + tagged field terminators).
|
|
e.Encode(APIVersionsResponse{APIKeys: apiKeys})
|
|
} else {
|
|
// Non-flexible: int32 array counts, int16 strings, no tagged fields
|
|
e.PutInt16(0) // error_code
|
|
e.PutArrayLen(len(apiKeys))
|
|
for _, k := range apiKeys {
|
|
e.PutInt16(k.APIKey)
|
|
e.PutInt16(k.MinVersion)
|
|
e.PutInt16(k.MaxVersion)
|
|
}
|
|
if req.RequestAPIVersion >= 1 {
|
|
e.PutInt32(0) // throttle_time_ms (v1+)
|
|
}
|
|
}
|
|
|
|
e.PutLen()
|
|
resp := e.Bytes()
|
|
hexLen := len(resp)
|
|
if hexLen > 60 {
|
|
hexLen = 60
|
|
}
|
|
log.Info("ApiVersions v%d response len=%d hex=%x", req.RequestAPIVersion, len(resp), resp[:hexLen])
|
|
return resp
|
|
}
|