use type ConsumerGroupInstanceId

weed/mq/sub_coordinator/consumer_group.go

@@ -106,7 +106,7 @@ func (cg *ConsumerGroup) BalanceConsumerGroupInstances(knownPartitionSlotToBroke
 	cg.mapping.BalanceToConsumerInstances(partitionSlotToBrokerList, consumerInstances)
 
 	// convert cg.mapping currentMapping to map of consumer group instance id to partition slots
-	consumerInstanceToPartitionSlots := make(map[string][]*PartitionSlotToConsumerInstance)
+	consumerInstanceToPartitionSlots := make(map[ConsumerGroupInstanceId][]*PartitionSlotToConsumerInstance)
 	for _, partitionSlot := range cg.mapping.currentMapping.PartitionSlots {
 		consumerInstanceToPartitionSlots[partitionSlot.AssignedInstanceId] = append(consumerInstanceToPartitionSlots[partitionSlot.AssignedInstanceId], partitionSlot)
 	}
@@ -117,7 +117,6 @@ func (cg *ConsumerGroup) BalanceConsumerGroupInstances(knownPartitionSlotToBroke
 		if !found {
 			partitionSlots = make([]*PartitionSlotToConsumerInstance, 0)
 		}
-		consumerGroupInstance.Partitions = ToPartitions(partitionSlotToBrokerList.RingSize, partitionSlots)
 		for _, partitionSlot := range partitionSlots {
 			consumerGroupInstance.ResponseChan <- &mq_pb.SubscriberToSubCoordinatorResponse{
 				Message: &mq_pb.SubscriberToSubCoordinatorResponse_Assignment_{

weed/mq/sub_coordinator/consumer_group_instance.go

@@ -6,17 +6,18 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/pb/mq_pb"
 )
 
+type ConsumerGroupInstanceId string
+
 type ConsumerGroupInstance struct {
-	InstanceId string
-	// the consumer group instance may not have an active partition
-	Partitions        []*topic.Partition
-	ResponseChan      chan *mq_pb.SubscriberToSubCoordinatorResponse
-	MaxPartitionCount int32
+	InstanceId         ConsumerGroupInstanceId
+	AssignedPartitions []topic.Partition
+	ResponseChan       chan *mq_pb.SubscriberToSubCoordinatorResponse
+	MaxPartitionCount  int32
 }
 
 func NewConsumerGroupInstance(instanceId string) *ConsumerGroupInstance {
 	return &ConsumerGroupInstance{
-		InstanceId:   instanceId,
+		InstanceId:   ConsumerGroupInstanceId(instanceId),
 		ResponseChan: make(chan *mq_pb.SubscriberToSubCoordinatorResponse, 1),
 	}
 }

weed/mq/sub_coordinator/partition_consumer_mapping.go

@@ -47,7 +47,7 @@ func (pcm *PartitionConsumerMapping) BalanceToConsumerInstances(partitionSlotToB
 
 func doBalanceSticky(partitions []*pub_balancer.PartitionSlotToBroker, consumerInstances []*ConsumerGroupInstance, prevMapping *PartitionSlotToConsumerInstanceList) (partitionSlots []*PartitionSlotToConsumerInstance) {
 	// collect previous consumer instance ids
-	prevConsumerInstanceIds := make(map[string]struct{})
+	prevConsumerInstanceIds := make(map[ConsumerGroupInstanceId]struct{})
 	if prevMapping != nil {
 		for _, prevPartitionSlot := range prevMapping.PartitionSlots {
 			if prevPartitionSlot.AssignedInstanceId != "" {
@@ -56,13 +56,13 @@ func doBalanceSticky(partitions []*pub_balancer.PartitionSlotToBroker, consumerI
 		}
 	}
 	// collect current consumer instance ids
-	currConsumerInstanceIds := make(map[string]struct{})
+	currConsumerInstanceIds := make(map[ConsumerGroupInstanceId]struct{})
 	for _, consumerInstance := range consumerInstances {
 		currConsumerInstanceIds[consumerInstance.InstanceId] = struct{}{}
 	}
 
 	// check deleted consumer instances
-	deletedConsumerInstanceIds := make(map[string]struct{})
+	deletedConsumerInstanceIds := make(map[ConsumerGroupInstanceId]struct{})
 	for consumerInstanceId := range prevConsumerInstanceIds {
 		if _, ok := currConsumerInstanceIds[consumerInstanceId]; !ok {
 			deletedConsumerInstanceIds[consumerInstanceId] = struct{}{}
@@ -100,7 +100,7 @@ func doBalanceSticky(partitions []*pub_balancer.PartitionSlotToBroker, consumerI
 
 	// for all consumer instances, count the average number of partitions
 	// that are assigned to them
-	consumerInstancePartitionCount := make(map[string]int)
+	consumerInstancePartitionCount := make(map[ConsumerGroupInstanceId]int)
 	for _, newPartitionSlot := range newPartitionSlots {
 		if newPartitionSlot.AssignedInstanceId != "" {
 			consumerInstancePartitionCount[newPartitionSlot.AssignedInstanceId]++

weed/mq/sub_coordinator/partition_list.go

@@ -7,7 +7,7 @@ type PartitionSlotToConsumerInstance struct {
 	RangeStop          int32
 	UnixTimeNs         int64
 	Broker             string
-	AssignedInstanceId string
+	AssignedInstanceId ConsumerGroupInstanceId
 	FollowerBroker     string
 }