mirror of
https://github.com/hanzoai/stream.git
synced 2026-10-30 18:21:09 +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]>
113 lines
3.6 KiB
Go
113 lines
3.6 KiB
Go
package protocol
|
|
|
|
import (
|
|
log "github.com/hanzoai/kafka/logging"
|
|
"github.com/hanzoai/kafka/serde"
|
|
"github.com/hanzoai/kafka/types"
|
|
)
|
|
|
|
// FindCoordinatorRequestV4Plus represents the request for coordinator finding
|
|
type FindCoordinatorRequestV4Plus struct {
|
|
KeyType uint8
|
|
CoordinatorKeys []string
|
|
}
|
|
|
|
// FindCoordinatorResponse represents the response for a coordinator finding request.
|
|
type FindCoordinatorResponse struct {
|
|
ThrottleTimeMs uint32
|
|
Coordinators []FindCoordinatorResponseCoordinator
|
|
}
|
|
|
|
// FindCoordinatorResponseCoordinator represents a coordinator.
|
|
type FindCoordinatorResponseCoordinator struct {
|
|
Key string `kafka:"CompactString"`
|
|
NodeID uint32
|
|
Host string `kafka:"CompactString"`
|
|
Port uint32
|
|
ErrorCode uint16
|
|
ErrorMessage string `kafka:"CompactString"`
|
|
}
|
|
|
|
func (b *Broker) getFindCoordinatorResponse(req types.Request) []byte {
|
|
log.Info("getFindCoordinatorResponse v%d %v", req.RequestAPIVersion, req.Body)
|
|
decoder := serde.NewDecoder(req.Body)
|
|
var key string
|
|
if req.RequestAPIVersion == 0 {
|
|
// v0: Key (STRING only, no KeyType)
|
|
key = decoder.NullableString()
|
|
log.Info("FindCoordinatorRequest v0 key=%q", key)
|
|
} else if req.RequestAPIVersion < 3 {
|
|
// v1-v2: non-flexible, Key (STRING) + KeyType (INT8)
|
|
key = decoder.NullableString()
|
|
_ = decoder.UInt8() // keyType (0=group, 1=txn)
|
|
log.Info("FindCoordinatorRequest v%d key=%q", req.RequestAPIVersion, key)
|
|
} else if req.RequestAPIVersion == 3 {
|
|
// v3: flexible, Key (COMPACT_STRING) + KeyType (INT8)
|
|
key = decoder.CompactString()
|
|
_ = decoder.UInt8() // keyType
|
|
decoder.EndStruct() // tagged fields
|
|
log.Info("FindCoordinatorRequest v3 key=%q", key)
|
|
} else {
|
|
// v4+: flexible, KeyType (INT8) + CoordinatorKeys (COMPACT_ARRAY)
|
|
findCoordinatorRequest := decoder.Decode(&FindCoordinatorRequestV4Plus{}).(*FindCoordinatorRequestV4Plus)
|
|
log.Info("FindCoordinatorRequestV4Plus %+v", findCoordinatorRequest)
|
|
|
|
if len(findCoordinatorRequest.CoordinatorKeys) > 0 {
|
|
key = findCoordinatorRequest.CoordinatorKeys[0]
|
|
}
|
|
}
|
|
|
|
return encodeFindCoordinatorResponse(req, key, b.Config.NodeID, b.Config.BrokerHost, uint32(b.Config.BrokerPort))
|
|
}
|
|
|
|
// encodeFindCoordinatorResponse encodes the FindCoordinator response based on API version.
|
|
// v0: error_code, node_id, host, port (flat, no throttle)
|
|
// v1-v2: throttle_time_ms, error_code, error_message, node_id, host, port (flat, non-flexible)
|
|
// v3: same as v1-v2 but flexible encoding
|
|
// v4+: flexible with Coordinators array
|
|
func encodeFindCoordinatorResponse(req types.Request, key string, nodeID int, host string, port uint32) []byte {
|
|
e := serde.NewEncoder()
|
|
e.PutInt32(req.CorrelationID)
|
|
|
|
if req.RequestAPIVersion >= 3 {
|
|
// Flexible response header
|
|
e.EndStruct()
|
|
if req.RequestAPIVersion >= 4 {
|
|
// v4+: Coordinators array (flexible)
|
|
response := FindCoordinatorResponse{
|
|
Coordinators: []FindCoordinatorResponseCoordinator{{
|
|
Key: key,
|
|
NodeID: uint32(nodeID),
|
|
Host: host,
|
|
Port: port,
|
|
}},
|
|
}
|
|
e.Encode(response)
|
|
} else {
|
|
// v3: flat response but flexible encoding
|
|
e.PutInt32(0) // throttle_time_ms
|
|
e.PutInt16(0) // error_code
|
|
e.PutCompactString("") // error_message (nullable)
|
|
e.PutInt32(uint32(nodeID))
|
|
e.PutCompactString(host)
|
|
e.PutInt32(port)
|
|
e.EndStruct() // response tagged fields
|
|
}
|
|
} else {
|
|
// Non-flexible (v0-v2)
|
|
if req.RequestAPIVersion >= 1 {
|
|
e.PutInt32(0) // throttle_time_ms
|
|
}
|
|
e.PutInt16(0) // error_code
|
|
if req.RequestAPIVersion >= 1 {
|
|
e.PutNullableString("") // error_message (null)
|
|
}
|
|
e.PutInt32(uint32(nodeID))
|
|
e.PutString(host)
|
|
e.PutInt32(port)
|
|
}
|
|
|
|
e.PutLen()
|
|
return e.Bytes()
|
|
}
|