Zach Kelling
74858afc40
- Module path: github.com/hanzoai/kafka -> github.com/hanzoai/stream - Docker image: ghcr.io/hanzoai/stream - Binary: hanzo-stream - All imports updated across 20+ Go files
104 lines
3.2 KiB
Go
104 lines
3.2 KiB
Go
package protocol
|
|
|
|
import (
|
|
log "github.com/hanzoai/stream/logging"
|
|
"github.com/hanzoai/stream/serde"
|
|
"github.com/hanzoai/stream/types"
|
|
"github.com/nats-io/nats.go"
|
|
)
|
|
|
|
// CreateTopicsResponse represents the response to a topic creation request.
|
|
type CreateTopicsResponse struct {
|
|
ThrottleTimeMs uint32
|
|
Topics []CreateTopicsResponseTopic
|
|
}
|
|
|
|
// CreateTopicsResponseTopic represents a topic's creation result.
|
|
type CreateTopicsResponseTopic struct {
|
|
Name string `kafka:"CompactString"`
|
|
TopicID [16]byte
|
|
ErrorCode uint16
|
|
ErrorMessage string `kafka:"CompactString"`
|
|
NumPartitions uint32
|
|
ReplicationFactor uint16
|
|
Configs []CreateTopicsResponseConfig
|
|
}
|
|
|
|
// CreateTopicsRequest represents the Kafka request to create topics.
|
|
type CreateTopicsRequest struct {
|
|
Topics []CreateTopicsRequestTopic
|
|
TimeoutMs uint32
|
|
ValidateOnly bool
|
|
}
|
|
|
|
// CreateTopicsRequestTopic represents the details of a topic to be created.
|
|
type CreateTopicsRequestTopic struct {
|
|
Name string `kafka:"CompactString"`
|
|
NumPartitions uint32
|
|
ReplicationFactor uint16
|
|
Assignments []CreateTopicsRequestAssignment
|
|
Configs []CreateTopicsRequestConfig
|
|
}
|
|
|
|
// CreateTopicsRequestAssignment represents the partition assignments for a topic.
|
|
type CreateTopicsRequestAssignment struct {
|
|
PartitionIndex uint32
|
|
BrokerIds []uint32
|
|
}
|
|
|
|
// CreateTopicsRequestConfig represents the configuration for a topic.
|
|
type CreateTopicsRequestConfig struct {
|
|
Name string `kafka:"CompactString"`
|
|
Value string `kafka:"CompactNullableString"`
|
|
}
|
|
|
|
// CreateTopicsResponseConfig represents a configuration for a topic.
|
|
type CreateTopicsResponseConfig struct {
|
|
Name string
|
|
Value string
|
|
ReadOnly bool
|
|
ConfigSource uint8
|
|
IsSensitive bool
|
|
}
|
|
|
|
// CreateTopics (Api key = 19)
|
|
func (b *Broker) getCreateTopicResponse(req types.Request) []byte {
|
|
decoder := serde.NewDecoder(req.Body)
|
|
createTopicsRequest := decoder.Decode(&CreateTopicsRequest{}).(*CreateTopicsRequest)
|
|
log.Debug("CreateTopicsRequest %+v", createTopicsRequest)
|
|
response := CreateTopicsResponse{}
|
|
|
|
for _, topic := range createTopicsRequest.Topics {
|
|
if int32(topic.NumPartitions) == -1 {
|
|
topic.NumPartitions = DefaultNumPartition
|
|
}
|
|
topicResponse := CreateTopicsResponseTopic{
|
|
Name: topic.Name,
|
|
TopicID: [16]byte{},
|
|
NumPartitions: topic.NumPartitions,
|
|
ReplicationFactor: topic.ReplicationFactor,
|
|
Configs: []CreateTopicsResponseConfig{},
|
|
}
|
|
|
|
if b.PubSub.TopicExists(topic.Name) {
|
|
topicResponse.ErrorCode = uint16(ErrTopicAlreadyExists.Code)
|
|
topicResponse.ErrorMessage = ErrTopicAlreadyExists.Message
|
|
} else {
|
|
replicas := b.Config.StreamReplicas
|
|
if replicas < 1 {
|
|
replicas = 1
|
|
}
|
|
err := b.PubSub.CreateTopicStreams(topic.Name, topic.NumPartitions, replicas, nats.FileStorage)
|
|
if err != nil {
|
|
log.Error("Error creating topic streams: %v", err)
|
|
topicResponse.ErrorCode = uint16(ErrUnknownServerError.Code)
|
|
topicResponse.ErrorMessage = err.Error()
|
|
}
|
|
}
|
|
response.Topics = append(response.Topics, topicResponse)
|
|
}
|
|
log.Debug("CreateTopicResponse %+v", response)
|
|
encoder := serde.NewEncoder()
|
|
return encoder.EncodeResponseBytes(req, response)
|
|
}
|