mirror of
https://github.com/hanzoai/stream.git
synced 2026-10-30 18:21:09 +00:00
zeekayandHanzo Dev
c3e231795c
This is the Kafka interface to Hanzo PubSub, not a product of its own. "Stream"
described the transport rather than the contract, and left two names for one
thing -- the repo was already redirecting hanzoai/kafka -> hanzoai/stream, so
imports and docs disagreed with the canonical URL.
The layering it belongs to:
hanzoai/tasks workloads and async workflows
|
hanzoai/pubsub the ONE event bus (JetStream-durable NATS)
|
hanzoai/kafka Kafka-wire adaptor (AMQP is the sibling seam)
|
hanzoai/datastore the ONE warehouse
Module path is now github.com/hanzoai/kafka; binary, image labels and docs say
Hanzo Kafka.
The word "stream" survives wherever it names the PROTOCOL rather than the
product -- JetStream stream names, ParseStreamName, the subject patterns in
pubsub/streams.go. Renaming those would have broken the wire contract to make a
brand consistent.
Unit tests pass. test/cluster and test/e2e still fail identically to before this
change: they shell out to an Apache Kafka distribution at ~/kafka_2.13-3.9.0/
that is not installed here, so they are unrelated to the rename.
Co-authored-by: Hanzo Dev <[email protected]>
103 lines
3.5 KiB
Go
103 lines
3.5 KiB
Go
package protocol
|
|
|
|
import (
|
|
"strconv"
|
|
|
|
log "github.com/hanzoai/kafka/logging"
|
|
"github.com/hanzoai/kafka/serde"
|
|
"github.com/hanzoai/kafka/types"
|
|
)
|
|
|
|
// ResourceType represents the kafka resource type (topic, group, etc.)
|
|
type ResourceType int8
|
|
|
|
const (
|
|
ResourceTypeUnknown ResourceType = 0
|
|
ResourceTypeAny ResourceType = 1
|
|
ResourceTypeTopic ResourceType = 2
|
|
ResourceTypeGroup ResourceType = 3
|
|
ResourceTypeBroker ResourceType = 4
|
|
ResourceTypeCluster ResourceType = 4
|
|
ResourceTypeTransactionalID ResourceType = 5
|
|
ResourceTypeDelegationToken ResourceType = 6
|
|
)
|
|
|
|
// DescribeConfigsRequest struct represents the request for DescribeConfigs with version 4.
|
|
type DescribeConfigsRequest struct {
|
|
Resources []DescribeConfigsResource
|
|
IncludeSynonyms bool
|
|
IncludeDocumentation bool
|
|
}
|
|
|
|
// DescribeConfigsResource struct represents the resource inside DescribeConfigsRequest.
|
|
type DescribeConfigsResource struct {
|
|
ResourceType uint8
|
|
ResourceName string `kafka:"CompactString"`
|
|
ConfigurationKeys []string `kafka:"CompactString"`
|
|
}
|
|
|
|
// DescribeConfigsResponse struct represents the response for DescribeConfigs with version 4.
|
|
type DescribeConfigsResponse struct {
|
|
ThrottleTimeMs uint32
|
|
Results []DescribeConfigsResponseResult
|
|
}
|
|
|
|
// DescribeConfigsResponseResult struct represents the result inside DescribeConfigsResponse.
|
|
type DescribeConfigsResponseResult struct {
|
|
ErrorCode uint16
|
|
ErrorMessage string `kafka:"CompactString"`
|
|
ResourceType uint8
|
|
ResourceName string `kafka:"CompactString"`
|
|
Configs []DescribeConfigsResponseConfig
|
|
}
|
|
|
|
// DescribeConfigsResponseConfig struct represents the configuration details within a result.
|
|
type DescribeConfigsResponseConfig struct {
|
|
Name string
|
|
Value string `kafka:"CompactNullableString"`
|
|
ReadOnly bool
|
|
ConfigSource uint8
|
|
IsSensitive bool
|
|
Synonyms []DescribeConfigsResponseSynonym
|
|
ConfigType uint
|
|
Documentation string `kafka:"CompactNullableString"`
|
|
}
|
|
|
|
// DescribeConfigsResponseSynonym struct represents synonym details within a config.
|
|
type DescribeConfigsResponseSynonym struct {
|
|
Name string `kafka:"CompactString"`
|
|
Value string `kafka:"CompactNullableString"`
|
|
Source uint
|
|
}
|
|
|
|
// DescribeConfigs (Api key = 32)
|
|
func (b *Broker) getDescribeConfigsResponse(req types.Request) []byte {
|
|
decoder := serde.NewDecoder(req.Body)
|
|
describeConfigsRequest := decoder.Decode(&DescribeConfigsRequest{}).(*DescribeConfigsRequest)
|
|
log.Debug("DescribeConfigsRequest %+v", describeConfigsRequest)
|
|
response := DescribeConfigsResponse{}
|
|
|
|
for _, resourceConfReq := range describeConfigsRequest.Resources {
|
|
resourceResp := DescribeConfigsResponseResult{
|
|
ResourceType: resourceConfReq.ResourceType,
|
|
ResourceName: resourceConfReq.ResourceName,
|
|
}
|
|
if ResourceType(resourceConfReq.ResourceType) == ResourceTypeTopic {
|
|
numPartitions, err := b.PubSub.GetTopicPartitionCount(resourceConfReq.ResourceName)
|
|
if err != nil || numPartitions == 0 {
|
|
resourceResp.ErrorCode = uint16(ErrUnknownTopicOrPartition.Code)
|
|
resourceResp.ErrorMessage = ErrUnknownTopicOrPartition.Message
|
|
} else {
|
|
resourceResp.Configs = append(resourceResp.Configs, DescribeConfigsResponseConfig{
|
|
Name: "partitions",
|
|
Value: strconv.Itoa(int(numPartitions)),
|
|
})
|
|
}
|
|
}
|
|
response.Results = append(response.Results, resourceResp)
|
|
}
|
|
log.Debug("DescribeConfigsResponse %+v", response)
|
|
encoder := serde.NewEncoder()
|
|
return encoder.EncodeResponseBytes(req, response)
|
|
}
|