mirror/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2024-11-28 05:15:00 +08:00
Code Issues Packages Projects Releases Wiki Activity

lookup existing topic partitions

Browse Source
This commit is contained in:
chrislu 2024-01-02 17:29:35 -08:00
parent 54e9877093
commit efb695fd93
2 changed files with 3 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
weed/mq/broker/broker_grpc_lookup.go
View File

@ -14,7 +14,6 @@ import (
//
// 2. find the topic partitions on the filer
// 2.1 if the topic is not found, return error
// 2.1.1 if the request is_for_subscribe, return error not found
// 2.1.2 if the request is_for_publish, create the topic
// 2.2 if the topic is found, return the brokers
//
@ -36,7 +35,7 @@ func (b *MessageQueueBroker) LookupTopicBrokers(ctx context.Context, request *mq
ret := &mq_pb.LookupTopicBrokersResponse{}
ret.Topic = request.Topic
ret.BrokerPartitionAssignments, err = b.Balancer.LookupOrAllocateTopicPartitions(ret.Topic, request.IsForPublish, 6)
ret.BrokerPartitionAssignments, err = b.Balancer.LookupOrAllocateTopicPartitions(ret.Topic, request.IsForPublish, -1)
return ret, err
}

5
weed/mq/pub_balancer/lookup.go
View File

@ -2,7 +2,6 @@ package pub_balancer
import (
"errors"
"github.com/seaweedfs/seaweedfs/weed/glog"
"github.com/seaweedfs/seaweedfs/weed/pb/mq_pb"
)
@ -34,8 +33,8 @@ func (balancer *Balancer) LookupOrAllocateTopicPartitions(topic *mq_pb.Topic, pu
}
}
}
if len(assignments) > 0 && len(assignments) == int(partitionCount) || !publish {
glog.V(0).Infof("existing topic partitions %d: %v", len(assignments), assignments)
if len(assignments) > 0 || !publish {
// glog.V(0).Infof("existing topic partitions %d: %v", len(assignments), assignments)
return assignments, nil
}