From 7121ae9617b76aa5028dbdfcd833ca8330fc212e Mon Sep 17 00:00:00 2001
From: chrislu <chris.lu@gmail.com>
Date: Mon, 22 Jan 2024 00:51:31 -0800
Subject: [PATCH] separate out read topic conf and ensure topic active
 assignments

---
 weed/mq/broker/broker_grpc_configure.go       | 27 +++++++++++--------
 .../mq/broker/broker_topic_conf_read_write.go | 12 ++++++---
 2 files changed, 25 insertions(+), 14 deletions(-)

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
 }