Browse Source

read existing topic conf

pull/5637/head
chrislu 11 months ago
parent
commit
9e6ea80dff
  1. 9
      weed/mq/broker/broker_grpc_configure.go

9
weed/mq/broker/broker_grpc_configure.go

@ -30,16 +30,17 @@ func (b *MessageQueueBroker) ConfigureTopic(ctx context.Context, request *mq_pb.
} }
t := topic.FromPbTopic(request.Topic) t := topic.FromPbTopic(request.Topic)
resp, err = b.readTopicConfFromFiler(t)
if err != nil {
var readErr error
resp, readErr = b.readTopicConfFromFiler(t)
if readErr != nil {
glog.V(0).Infof("read topic %s conf: %v", request.Topic, err) glog.V(0).Infof("read topic %s conf: %v", request.Topic, err)
} else { } else {
err = b.ensureTopicActiveAssignments(t, resp)
readErr = b.ensureTopicActiveAssignments(t, resp)
// no need to assign directly. // no need to assign directly.
// The added or updated assignees will read from filer directly. // The added or updated assignees will read from filer directly.
// The gone assignees will die by themselves. // The gone assignees will die by themselves.
} }
if err == nil && len(resp.BrokerPartitionAssignments) == int(request.PartitionCount) {
if readErr == nil && len(resp.BrokerPartitionAssignments) == int(request.PartitionCount) {
glog.V(0).Infof("existing topic partitions %d: %+v", len(resp.BrokerPartitionAssignments), resp.BrokerPartitionAssignments) glog.V(0).Infof("existing topic partitions %d: %+v", len(resp.BrokerPartitionAssignments), resp.BrokerPartitionAssignments)
} else { } else {
if resp!=nil && len(resp.BrokerPartitionAssignments) > 0 { if resp!=nil && len(resp.BrokerPartitionAssignments) > 0 {

Loading…
Cancel
Save