stream/protocol/metadata.go

package protocol

import (
	log "github.com/hanzoai/kafka/logging"
	"github.com/hanzoai/kafka/serde"
	"github.com/hanzoai/kafka/types"
	"github.com/nats-io/nats.go"
)

// Metadata (Api key = 3)

// MetadataResponseBroker represents a broker in a metadata response.
type MetadataResponseBroker struct {
	NodeID uint32
	Host   string `kafka:"CompactString"`
	Port   uint32
	Rack   string `kafka:"CompactString"`
}

// MetadataResponsePartition represents partition information in a metadata response.
type MetadataResponsePartition struct {
	ErrorCode       uint16
	PartitionIndex  uint32
	LeaderID        uint32
	LeaderEpoch     uint32
	ReplicaNodes    []uint32
	IsrNodes        []uint32
	OfflineReplicas []uint32
}

// MetadataRequest represents a metadata request.
type MetadataRequest struct {
	Topics                           []MetadataRequestTopic
	AllowAutoTopicCreation           bool
	IncludeTopicAuthorizedOperations bool
}

// MetadataRequestTopic represents a topic in the metadata request.
type MetadataRequestTopic struct {
	TopicID [16]byte
	Name    string `kafka:"CompactString"`
}

// MetadataResponseTopic represents a topic in the metadata response.
type MetadataResponseTopic struct {
	ErrorCode                 uint16
	Name                      string `kafka:"CompactString"`
	TopicID                   [16]byte
	IsInternal                bool
	Partitions                []MetadataResponsePartition
	TopicAuthorizedOperations uint32
}

// MetadataResponse represents a metadata response with brokers, topics, and more.
type MetadataResponse struct {
	ThrottleTimeMs uint32
	Brokers        []MetadataResponseBroker
	ClusterID      string `kafka:"CompactString"` // nullable
	ControllerID   uint32
	Topics         []MetadataResponseTopic
}

func decodeMetadataRequest(d serde.Decoder, req *MetadataRequest, apiVersion uint16) {
	if apiVersion >= 9 {
		// Flexible version: compact arrays, compact strings, tagged fields
		lenTopics := int(d.CompactArrayLen())
		for range lenTopics {
			topic := MetadataRequestTopic{
				TopicID: d.UUID(),
				Name:    d.CompactString(),
			}
			req.Topics = append(req.Topics, topic)
			d.EndStruct() // topic tagged fields
		}
		req.AllowAutoTopicCreation = d.Bool()
		req.IncludeTopicAuthorizedOperations = d.Bool()
		// request tagged fields consumed by caller if needed
	} else {
		// Non-flexible: int32 array len, int16 strings, no tagged fields
		arrayLen := int(int32(d.UInt32()))
		if arrayLen > 0 {
			for range arrayLen {
				topic := MetadataRequestTopic{Name: d.NullableString()}
				req.Topics = append(req.Topics, topic)
			}
		}
		if apiVersion >= 4 {
			req.AllowAutoTopicCreation = d.Bool()
		}
	}
}

func (b *Broker) getMetadataResponse(req types.Request) []byte {
	decoder := serde.NewDecoder(req.Body)
	metadataRequest := &MetadataRequest{}
	decodeMetadataRequest(decoder, metadataRequest, req.RequestAPIVersion)
	log.Debug("metadataRequest %+v", metadataRequest)

	nodeID := uint32(b.Config.NodeID)
	brokers := []MetadataResponseBroker{{
		NodeID: nodeID,
		Host:   b.Config.BrokerHost,
		Port:   uint32(b.Config.BrokerPort),
	}}

	// If no topics requested, list all
	if len(metadataRequest.Topics) == 0 {
		topicNames, _ := b.PubSub.ListTopics()
		for _, t := range topicNames {
			metadataRequest.Topics = append(metadataRequest.Topics, MetadataRequestTopic{Name: t})
		}
	}

	var topics []MetadataResponseTopic
	for _, reqTopic := range metadataRequest.Topics {
		topic := MetadataResponseTopic{Name: reqTopic.Name, TopicID: reqTopic.TopicID, IsInternal: false}

		if b.PubSub.TopicExists(reqTopic.Name) {
			numPartitions, _ := b.PubSub.GetTopicPartitionCount(reqTopic.Name)
			for i := range numPartitions {
				topic.Partitions = append(topic.Partitions, MetadataResponsePartition{
					PartitionIndex: i,
					LeaderID:       nodeID,
					ReplicaNodes:   []uint32{nodeID},
					IsrNodes:       []uint32{nodeID},
				})
			}
		} else if metadataRequest.AllowAutoTopicCreation {
			replicas := max(b.Config.StreamReplicas, 1)
			b.PubSub.CreateTopicStreams(reqTopic.Name, 1, replicas, nats.FileStorage)
			topic.Partitions = append(topic.Partitions, MetadataResponsePartition{
				PartitionIndex: 0,
				LeaderID:       nodeID,
				ReplicaNodes:   []uint32{nodeID},
				IsrNodes:       []uint32{nodeID},
			})
		} else {
			topic.ErrorCode = uint16(ErrUnknownTopicOrPartition.Code)
		}

		topics = append(topics, topic)
	}

	response := MetadataResponse{
		Brokers:      brokers,
		ClusterID:    ClusterID,
		ControllerID: nodeID,
		Topics:       topics,
	}
	log.Debug("MetadataResponse %+v", response)
	return encodeMetadataResponse(req, response)
}

// encodeMetadataResponse manually encodes the Metadata response based on API version.
// v0-v8: non-flexible (int32 arrays, int16 strings, no tagged fields).
// v9+: flexible (compact arrays/strings, tagged field terminators).
func encodeMetadataResponse(req types.Request, response MetadataResponse) []byte {
	e := serde.NewEncoder()
	e.PutInt32(req.CorrelationID)

	if req.RequestAPIVersion >= 9 {
		// Flexible response
		e.EndStruct()
		e.Encode(response)
	} else {
		// Non-flexible response
		if req.RequestAPIVersion >= 3 {
			e.PutInt32(response.ThrottleTimeMs)
		}
		// brokers array
		e.PutArrayLen(len(response.Brokers))
		for _, broker := range response.Brokers {
			e.PutInt32(broker.NodeID)
			e.PutString(broker.Host)
			e.PutInt32(broker.Port)
			if req.RequestAPIVersion >= 1 {
				e.PutNullableString(broker.Rack)
			}
		}
		if req.RequestAPIVersion >= 2 {
			e.PutNullableString(response.ClusterID)
		}
		if req.RequestAPIVersion >= 1 {
			e.PutInt32(response.ControllerID)
		}
		// topics array
		e.PutArrayLen(len(response.Topics))
		for _, topic := range response.Topics {
			e.PutInt16(topic.ErrorCode)
			e.PutString(topic.Name)
			if req.RequestAPIVersion >= 1 {
				e.PutBool(topic.IsInternal)
			}
			// partitions array
			e.PutArrayLen(len(topic.Partitions))
			for _, p := range topic.Partitions {
				e.PutInt16(p.ErrorCode)
				e.PutInt32(p.PartitionIndex)
				e.PutInt32(p.LeaderID)
				if req.RequestAPIVersion >= 7 {
					e.PutInt32(p.LeaderEpoch)
				}
				// replica nodes
				e.PutArrayLen(len(p.ReplicaNodes))
				for _, r := range p.ReplicaNodes {
					e.PutInt32(r)
				}
				// isr nodes
				e.PutArrayLen(len(p.IsrNodes))
				for _, r := range p.IsrNodes {
					e.PutInt32(r)
				}
				if req.RequestAPIVersion >= 5 {
					// offline replicas
					e.PutArrayLen(len(p.OfflineReplicas))
					for _, r := range p.OfflineReplicas {
						e.PutInt32(r)
					}
				}
			}
			if req.RequestAPIVersion >= 8 {
				e.PutInt32(topic.TopicAuthorizedOperations)
			}
		}
	}
	e.PutLen()
	return e.Bytes()
}