From bf13f3ced78fd1556f02bc4a28c0c342dc15a76f Mon Sep 17 00:00:00 2001 From: chrislu Date: Mon, 2 Oct 2023 01:02:27 -0700 Subject: [PATCH] add subscriber coordinator --- weed/mq/client/sub_client/subscribe.go | 5 ++ weed/mq/client/sub_client/subscriber.go | 12 ++-- weed/mq/coordinator/consumer_group.go | 92 +++++++++++++++++++++++++ weed/mq/coordinator/coordinator.go | 36 ++++++++++ weed/mq/topic/partition.go | 17 +++++ 5 files changed, 157 insertions(+), 5 deletions(-) create mode 100644 weed/mq/coordinator/consumer_group.go create mode 100644 weed/mq/coordinator/coordinator.go diff --git a/weed/mq/client/sub_client/subscribe.go b/weed/mq/client/sub_client/subscribe.go index bfde6a512..0803b2c79 100644 --- a/weed/mq/client/sub_client/subscribe.go +++ b/weed/mq/client/sub_client/subscribe.go @@ -11,9 +11,14 @@ import ( func (sub *TopicSubscriber) Subscribe() error { util.RetryUntil("subscribe", func() error { + // ask balancer for brokers of the topic if err := sub.doLookup(sub.bootstrapBroker); err != nil { return fmt.Errorf("lookup topic %s/%s: %v", sub.ContentConfig.Namespace, sub.ContentConfig.Topic, err) } + // treat the first broker as the topic leader + // connect to the leader broker + + // subscribe to the topic if err := sub.doProcess(); err != nil { return fmt.Errorf("subscribe topic %s/%s: %v", sub.ContentConfig.Namespace, sub.ContentConfig.Topic, err) } diff --git a/weed/mq/client/sub_client/subscriber.go b/weed/mq/client/sub_client/subscriber.go index f744c6fa2..809673de1 100644 --- a/weed/mq/client/sub_client/subscriber.go +++ b/weed/mq/client/sub_client/subscriber.go @@ -6,11 +6,13 @@ import ( ) type SubscriberConfiguration struct { - ClientId string - GroupId string - GroupInstanceId string - BootstrapServers []string - GrpcDialOption grpc.DialOption + ClientId string + GroupId string + GroupInstanceId string + GroupMinimumPeers int32 + GroupMaximumPeers int32 + BootstrapServers []string + GrpcDialOption grpc.DialOption } type ContentConfiguration struct { diff --git a/weed/mq/coordinator/consumer_group.go b/weed/mq/coordinator/consumer_group.go new file mode 100644 index 000000000..e3dec493c --- /dev/null +++ b/weed/mq/coordinator/consumer_group.go @@ -0,0 +1,92 @@ +package coordinator + +import ( + "github.com/seaweedfs/seaweedfs/weed/mq/topic" + "sync" +) + +func (cg *ConsumerGroup) SetMinMaxActiveInstances(min, max int32) { + cg.MinimumActiveInstances = min + cg.MaximumActiveInstances = max +} + +func (cg *ConsumerGroup) AddConsumerGroupInstance(clientId string) *ConsumerGroupInstance { + cgi := &ConsumerGroupInstance{ + ClientId: clientId, + } + cg.ConsumerGroupInstances.Set(clientId, cgi) + return cgi +} + +func (cg *ConsumerGroup) RemoveConsumerGroupInstance(clientId string) { + cg.ConsumerGroupInstances.Remove(clientId) +} + +func (cg *ConsumerGroup) CoordinateIfNeeded() { + emptyInstanceCount, activeInstanceCount := int32(0), int32(0) + for cgi := range cg.ConsumerGroupInstances.IterBuffered() { + if cgi.Val.Partition == nil { + // this consumer group instance is not assigned a partition + // need to assign one + emptyInstanceCount++ + } else { + activeInstanceCount++ + } + } + + var delta int32 + if emptyInstanceCount > 0 { + if cg.MinimumActiveInstances <= 0 { + // need to assign more partitions + delta = emptyInstanceCount + } else if activeInstanceCount < cg.MinimumActiveInstances && activeInstanceCount+emptyInstanceCount >= cg.MinimumActiveInstances { + // need to assign more partitions + delta = cg.MinimumActiveInstances - activeInstanceCount + } + } + + if cg.MaximumActiveInstances > 0 { + if activeInstanceCount > cg.MaximumActiveInstances { + // need to remove some partitions + delta = cg.MaximumActiveInstances - activeInstanceCount + } + } + if delta == 0 { + return + } + cg.doCoordinate(activeInstanceCount + delta) +} + +func (cg *ConsumerGroup) doCoordinate(target int32) { + // stop existing instances from processing + var wg sync.WaitGroup + for cgi := range cg.ConsumerGroupInstances.IterBuffered() { + if cgi.Val.Partition != nil { + wg.Add(1) + go func(cgi *ConsumerGroupInstance) { + defer wg.Done() + // stop processing + // flush internal state + // wait for all messages to be processed + // close the connection + }(cgi.Val) + } + } + wg.Wait() + + partitions := topic.SplitPartitions(target) + + // assign partitions to new instances + i := 0 + for cgi := range cg.ConsumerGroupInstances.IterBuffered() { + cgi.Val.Partition = partitions[i] + i++ + wg.Add(1) + go func(cgi *ConsumerGroupInstance) { + defer wg.Done() + // start processing + // start consuming from the last offset + }(cgi.Val) + } + wg.Wait() +} diff --git a/weed/mq/coordinator/coordinator.go b/weed/mq/coordinator/coordinator.go new file mode 100644 index 000000000..e94ac3371 --- /dev/null +++ b/weed/mq/coordinator/coordinator.go @@ -0,0 +1,36 @@ +package coordinator + +import ( + cmap "github.com/orcaman/concurrent-map/v2" + "github.com/seaweedfs/seaweedfs/weed/mq/topic" +) + +type ConsumerGroupInstance struct { + ClientId string + // the consumer group instance may not have an active partition + Partition *topic.Partition + // processed message count + ProcessedMessageCount int64 +} +type ConsumerGroup struct { + // map a client id to a consumer group instance + ConsumerGroupInstances cmap.ConcurrentMap[string, *ConsumerGroupInstance] + MinimumActiveInstances int32 + MaximumActiveInstances int32 +} +type TopicConsumerGroups struct { + // map a consumer group name to a consumer group + ConsumerGroups cmap.ConcurrentMap[string, *ConsumerGroup] +} + +// Coordinator coordinates the instances in the consumer group for one topic. +// It is responsible for: +// 1. Assigning partitions to consumer instances. +// 2. Reassigning partitions when a consumer instance is down. +// 3. Reassigning partitions when a consumer instance is up. +type Coordinator struct { + // map client id to subscriber + Subscribers cmap.ConcurrentMap[string, *ConsumerGroupInstance] + // map topic name to consumer groups + TopicSubscribers cmap.ConcurrentMap[string, map[string]TopicConsumerGroups] +} diff --git a/weed/mq/topic/partition.go b/weed/mq/topic/partition.go index 285bdcb36..79c830f13 100644 --- a/weed/mq/topic/partition.go +++ b/weed/mq/topic/partition.go @@ -30,3 +30,20 @@ func FromPbPartition(partition *mq_pb.Partition) Partition { RingSize: partition.RingSize, } } + +func SplitPartitions(targetCount int32) []*Partition { + partitions := make([]*Partition, 0, targetCount) + partitionSize := PartitionCount / targetCount + for i := int32(0); i < targetCount; i++ { + partitionStop := (i + 1) * partitionSize + if i == targetCount-1 { + partitionStop = PartitionCount + } + partitions = append(partitions, &Partition{ + RangeStart: i * partitionSize, + RangeStop: partitionStop, + RingSize: PartitionCount, + }) + } + return partitions +}