diff --git a/weed/mq/broker/broker_grpc_configure.go b/weed/mq/broker/broker_grpc_configure.go index 9ba67f250..8ffe64d8d 100644 --- a/weed/mq/broker/broker_grpc_configure.go +++ b/weed/mq/broker/broker_grpc_configure.go @@ -37,26 +37,31 @@ func (b *MessageQueueBroker) ConfigureTopic(ctx context.Context, request *mq_pb. return resp, err } - ret := &mq_pb.ConfigureTopicResponse{} - existingAssignments := b.Balancer.LookupTopicPartitions(request.Topic) - if len(existingAssignments) == int(request.PartitionCount) { - glog.V(0).Infof("existing topic partitions %d: %+v", len(existingAssignments), existingAssignments) - ret.BrokerPartitionAssignments = existingAssignments + t := topic.FromPbTopic(request.Topic) + resp, err = b.readTopicConfFromFiler(t) + if err != nil { + glog.V(0).Infof("read topic %s conf: %v", request.Topic, err) } else { + err = b.ensureTopicActiveAssignments(t, resp) + } + if err == nil && len(resp.BrokerPartitionAssignments) == int(request.PartitionCount) { + glog.V(0).Infof("existing topic partitions %d: %+v", len(resp.BrokerPartitionAssignments), resp.BrokerPartitionAssignments) + } else { + resp = &mq_pb.ConfigureTopicResponse{} if b.Balancer.Brokers.IsEmpty() { return nil, status.Errorf(codes.Unavailable, pub_balancer.ErrNoBroker.Error()) } - ret.BrokerPartitionAssignments = pub_balancer.AllocateTopicPartitions(b.Balancer.Brokers, request.PartitionCount) + resp.BrokerPartitionAssignments = pub_balancer.AllocateTopicPartitions(b.Balancer.Brokers, request.PartitionCount) // save the topic configuration on filer - if err := b.saveTopicConfToFiler(request.Topic, ret); err != nil { + if err := b.saveTopicConfToFiler(request.Topic, resp); err != nil { return nil, fmt.Errorf("configure topic: %v", err) } - b.Balancer.OnPartitionChange(request.Topic, ret.BrokerPartitionAssignments) + b.Balancer.OnPartitionChange(request.Topic, resp.BrokerPartitionAssignments) } - for _, bpa := range ret.BrokerPartitionAssignments { + for _, bpa := range resp.BrokerPartitionAssignments { fmt.Printf("create topic %s partition %+v on %s\n", request.Topic, bpa.Partition, bpa.LeaderBroker) if doCreateErr := b.withBrokerClient(false, pb.ServerAddress(bpa.LeaderBroker), func(client mq_pb.SeaweedMessagingClient) error { _, doCreateErr := client.AssignTopicPartitions(ctx, &mq_pb.AssignTopicPartitionsRequest{ @@ -86,9 +91,9 @@ func (b *MessageQueueBroker) ConfigureTopic(ctx context.Context, request *mq_pb. } } - glog.V(0).Infof("ConfigureTopic: topic %s partition assignments: %v", request.Topic, ret.BrokerPartitionAssignments) + glog.V(0).Infof("ConfigureTopic: topic %s partition assignments: %v", request.Topic, resp.BrokerPartitionAssignments) - return ret, err + return resp, err } // AssignTopicPartitions Runs on the assigned broker, to execute the topic partition assignment diff --git a/weed/mq/broker/broker_topic_conf_read_write.go b/weed/mq/broker/broker_topic_conf_read_write.go index 6c9a38e6d..397e70fac 100644 --- a/weed/mq/broker/broker_topic_conf_read_write.go +++ b/weed/mq/broker/broker_topic_conf_read_write.go @@ -28,6 +28,8 @@ func (b *MessageQueueBroker) saveTopicConfToFiler(t *mq_pb.Topic, conf *mq_pb.Co return nil } +// readTopicConfFromFiler reads the topic configuration from filer +// this should only be run in broker leader, to ensure correct active broker list. func (b *MessageQueueBroker) readTopicConfFromFiler(t topic.Topic) (conf *mq_pb.ConfigureTopicResponse, err error) { glog.V(0).Infof("load conf for topic %v from filer", t) @@ -48,14 +50,18 @@ func (b *MessageQueueBroker) readTopicConfFromFiler(t topic.Topic) (conf *mq_pb. return nil, err } + return conf, nil +} + +func (b *MessageQueueBroker) ensureTopicActiveAssignments(t topic.Topic, conf *mq_pb.ConfigureTopicResponse) (err error) { // also fix assignee broker if invalid changedAssignments := pub_balancer.EnsureAssignmentsToActiveBrokers(b.Balancer.Brokers, conf.BrokerPartitionAssignments) if len(changedAssignments) > 0 { glog.V(0).Infof("topic %v partition assignments changed: %v", t, changedAssignments) if err = b.saveTopicConfToFiler(t.ToPbTopic(), conf); err != nil { - return nil, err + return err } } - - return conf, err + + return err }