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]>
65 lines
2.3 KiB
Go
65 lines
2.3 KiB
Go
package protocol
|
|
|
|
import "github.com/hanzoai/kafka/types"
|
|
|
|
// https://kafka.apache.org/protocol#protocol_api_keys
|
|
var produceKey = uint16(0)
|
|
var fetchKey = uint16(1)
|
|
var listOffsetsKey = uint16(2)
|
|
var metadataKey = uint16(3)
|
|
var offsetCommitKey = uint16(8)
|
|
var offsetFetchKey = uint16(9)
|
|
var findCoordinatorKey = uint16(10)
|
|
var joinGroupKey = uint16(11)
|
|
var heartbeatKey = uint16(12)
|
|
var leaveGroupKey = uint16(13)
|
|
var syncGroupKey = uint16(14)
|
|
var apiVersionKey = uint16(18)
|
|
var createTopicKey = uint16(19)
|
|
var initProducerIDKey = uint16(22)
|
|
var describeConfigsKey = uint16(32)
|
|
|
|
// APIKeyHandler represents a kafka api key with its handler
|
|
type APIKeyHandler struct {
|
|
Name string
|
|
Handler func(req types.Request) []byte
|
|
}
|
|
|
|
// APIDispatcher maps the Request key to its handler
|
|
func (b *Broker) APIDispatcher(requestAPIKey uint16) APIKeyHandler {
|
|
switch requestAPIKey {
|
|
case produceKey:
|
|
return APIKeyHandler{Name: "Produce", Handler: b.getProduceResponse}
|
|
case fetchKey:
|
|
return APIKeyHandler{Name: "Fetch", Handler: b.getFetchResponse}
|
|
case listOffsetsKey:
|
|
return APIKeyHandler{Name: "ListOffsets", Handler: b.getListOffsetsResponse}
|
|
case metadataKey:
|
|
return APIKeyHandler{Name: "Metadata", Handler: b.getMetadataResponse}
|
|
case offsetCommitKey:
|
|
return APIKeyHandler{Name: "OffsetCommit", Handler: b.getOffsetCommitResponse}
|
|
case offsetFetchKey:
|
|
return APIKeyHandler{Name: "OffsetFetch", Handler: b.getOffsetFetchResponse}
|
|
case findCoordinatorKey:
|
|
return APIKeyHandler{Name: "FindCoordinator", Handler: b.getFindCoordinatorResponse}
|
|
case joinGroupKey:
|
|
return APIKeyHandler{Name: "JoinGroup", Handler: b.getJoinGroupResponse}
|
|
case heartbeatKey:
|
|
return APIKeyHandler{Name: "Heartbeat", Handler: b.getHeartbeatResponse}
|
|
case leaveGroupKey:
|
|
return APIKeyHandler{Name: "LeaveGroup", Handler: b.getLeaveGroupResponse}
|
|
case syncGroupKey:
|
|
return APIKeyHandler{Name: "SyncGroup", Handler: b.getSyncGroupResponse}
|
|
case apiVersionKey:
|
|
return APIKeyHandler{Name: "APIVersion", Handler: b.getAPIVersionResponse}
|
|
case createTopicKey:
|
|
return APIKeyHandler{Name: "CreateTopic", Handler: b.getCreateTopicResponse}
|
|
case initProducerIDKey:
|
|
return APIKeyHandler{Name: "InitProducerID", Handler: b.getInitProducerIDResponse}
|
|
case describeConfigsKey:
|
|
return APIKeyHandler{Name: "DescribeConfigs", Handler: b.getDescribeConfigsResponse}
|
|
default:
|
|
return APIKeyHandler{}
|
|
}
|
|
}
|