gsh-digital-services/seaweedFS
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

implement create topic

Browse Source
This commit is contained in:
chrislu
2023-09-24 15:08:44 -07:00
parent 0361c321b4
commit d74348048a
4 changed files with 52 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
weed/mq/balancer/lookup.go
View File

@@ -1,10 +1,15 @@
package balancer
import (
"errors"
"github.com/seaweedfs/seaweedfs/weed/pb/mq_pb"
)
func (b *Balancer) LookupOrAllocateTopicPartitions(topic *mq_pb.Topic, publish bool) (assignments []*mq_pb.BrokerPartitionAssignment, err error) {
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
@@ -39,5 +44,8 @@ func (b *Balancer) LookupOrAllocateTopicPartitions(topic *mq_pb.Topic, publish b
// if the request is_for_subscribe
// return error not found
// t := topic.FromPbTopic(request.Topic)
if b.Brokers.IsEmpty() {
return nil, ErrNoBroker
}
return allocateTopicPartitions(b.Brokers, 6), nil
}