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

rename

Browse Source
This commit is contained in:
chrislu
2024-05-13 19:24:47 -07:00
parent 372bd8d71d
commit 844a1a0192
1 changed files with 3 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
weed/mq/sub_coordinator/consumer_group.go
View File

@@ -53,7 +53,7 @@ func (cg *ConsumerGroup) onConsumerGroupInstanceChange(reason string) {
cg.reBalanceTimer = nil cg.reBalanceTimer = nil
} }
cg.reBalanceTimer = time.AfterFunc(5*time.Second, func() { cg.reBalanceTimer = time.AfterFunc(5*time.Second, func() {
cg.RebalanceConsumberGroupInstances(nil, reason) cg.BalanceConsumerGroupInstances(nil, reason)
cg.reBalanceTimer = nil cg.reBalanceTimer = nil
}) })
} }
@@ -66,10 +66,10 @@ func (cg *ConsumerGroup) OnPartitionListChange(assignments []*mq_pb.BrokerPartit
for _, assignment := range assignments { for _, assignment := range assignments {
partitionSlotToBrokerList.AddBroker(assignment.Partition, assignment.LeaderBroker) partitionSlotToBrokerList.AddBroker(assignment.Partition, assignment.LeaderBroker)
} }
cg.RebalanceConsumberGroupInstances(partitionSlotToBrokerList, "partition list change") cg.BalanceConsumerGroupInstances(partitionSlotToBrokerList, "partition list change")
} }
func (cg *ConsumerGroup) RebalanceConsumberGroupInstances(knownPartitionSlotToBrokerList *pub_balancer.PartitionSlotToBrokerList, reason string) { func (cg *ConsumerGroup) BalanceConsumerGroupInstances(knownPartitionSlotToBrokerList *pub_balancer.PartitionSlotToBrokerList, reason string) {
glog.V(0).Infof("rebalance consumer group %s due to %s", cg.topic.String(), reason) glog.V(0).Infof("rebalance consumer group %s due to %s", cg.topic.String(), reason)
// collect current topic partitions // collect current topic partitions