seaweedfs/weed/mq/balancer/lookup.go

49 lines
1.5 KiB
Go
Raw Normal View History

package balancer
import (
2023-09-24 22:08:44 +00:00
"errors"
"github.com/seaweedfs/seaweedfs/weed/pb/mq_pb"
)
2023-09-24 22:08:44 +00:00
var (
ErrNoBroker = errors.New("no broker")
)
func (b *Balancer) LookupOrAllocateTopicPartitions(topic *mq_pb.Topic, publish bool, partitionCount int32) (assignments []*mq_pb.BrokerPartitionAssignment, err error) {
// find existing topic partition assignments
for brokerStatsItem := range b.Brokers.IterBuffered() {
broker, brokerStats := brokerStatsItem.Key, brokerStatsItem.Val
for topicPartitionStatsItem := range brokerStats.Stats.IterBuffered() {
topicPartitionStat := topicPartitionStatsItem.Val
if topicPartitionStat.TopicPartition.Namespace == topic.Namespace &&
2023-10-02 08:01:45 +00:00
topicPartitionStat.TopicPartition.Name == topic.Name {
assignment := &mq_pb.BrokerPartitionAssignment{
Partition: &mq_pb.Partition{
RingSize: MaxPartitionCount,
RangeStart: topicPartitionStat.RangeStart,
RangeStop: topicPartitionStat.RangeStop,
},
}
2023-09-25 06:05:41 +00:00
// TODO fix follower setting
assignment.LeaderBroker = broker
assignments = append(assignments, assignment)
}
}
}
if len(assignments) > 0 {
return assignments, nil
}
// find the topic partitions on the filer
// if the topic is not found
// if the request is_for_publish
// create the topic
// if the request is_for_subscribe
// return error not found
// t := topic.FromPbTopic(request.Topic)
2023-09-24 22:08:44 +00:00
if b.Brokers.IsEmpty() {
return nil, ErrNoBroker
}
2023-09-25 04:19:51 +00:00
return allocateTopicPartitions(b.Brokers, partitionCount), nil
}