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

separate out read topic conf and ensure topic active assignments

Browse Source
This commit is contained in:
chrislu
2024-01-22 00:51:31 -08:00
parent c77d35313e
commit 7121ae9617
2 changed files with 25 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
weed/mq/broker/broker_grpc_configure.go
View File

@@ -37,26 +37,31 @@ func (b *MessageQueueBroker) ConfigureTopic(ctx context.Context, request *mq_pb.
return resp, err return resp, err
} }
ret := &mq_pb.ConfigureTopicResponse{} t := topic.FromPbTopic(request.Topic)
existingAssignments := b.Balancer.LookupTopicPartitions(request.Topic) resp, err = b.readTopicConfFromFiler(t)
if len(existingAssignments) == int(request.PartitionCount) { if err != nil {
glog.V(0).Infof("existing topic partitions %d: %+v", len(existingAssignments), existingAssignments) glog.V(0).Infof("read topic %s conf: %v", request.Topic, err)
ret.BrokerPartitionAssignments = existingAssignments
} else { } 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() { if b.Balancer.Brokers.IsEmpty() {
return nil, status.Errorf(codes.Unavailable, pub_balancer.ErrNoBroker.Error()) 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 // 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) 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) 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 { if doCreateErr := b.withBrokerClient(false, pb.ServerAddress(bpa.LeaderBroker), func(client mq_pb.SeaweedMessagingClient) error {
_, doCreateErr := client.AssignTopicPartitions(ctx, &mq_pb.AssignTopicPartitionsRequest{ _, 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 // AssignTopicPartitions Runs on the assigned broker, to execute the topic partition assignment

12
weed/mq/broker/broker_topic_conf_read_write.go
View File

@@ -28,6 +28,8 @@ func (b *MessageQueueBroker) saveTopicConfToFiler(t *mq_pb.Topic, conf *mq_pb.Co
return nil 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) { 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) 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 nil, err
} }
return conf, nil
}
func (b *MessageQueueBroker) ensureTopicActiveAssignments(t topic.Topic, conf *mq_pb.ConfigureTopicResponse) (err error) {
// also fix assignee broker if invalid // also fix assignee broker if invalid
changedAssignments := pub_balancer.EnsureAssignmentsToActiveBrokers(b.Balancer.Brokers, conf.BrokerPartitionAssignments) changedAssignments := pub_balancer.EnsureAssignmentsToActiveBrokers(b.Balancer.Brokers, conf.BrokerPartitionAssignments)
if len(changedAssignments) > 0 { if len(changedAssignments) > 0 {
glog.V(0).Infof("topic %v partition assignments changed: %v", t, changedAssignments) glog.V(0).Infof("topic %v partition assignments changed: %v", t, changedAssignments)
if err = b.saveTopicConfToFiler(t.ToPbTopic(), conf); err != nil { if err = b.saveTopicConfToFiler(t.ToPbTopic(), conf); err != nil {
return nil, err return err
} }
} }
return conf, err return err
} }