consumer instance passing MaxPartitionCount to coordinator

weed/mq/client/sub_client/connect_to_sub_coordinator.go

@@ -51,6 +51,7 @@ func (sub *TopicSubscriber) doKeepConnectedToSubCoordinator() {
 							ConsumerGroup:           sub.SubscriberConfig.ConsumerGroup,
 							ConsumerGroupInstanceId: sub.SubscriberConfig.ConsumerGroupInstanceId,
 							Topic:                   sub.ContentConfig.Topic.ToPbTopic(),
+							MaxPartitionCount: 	 	 sub.ProcessorConfig.ConcurrentPartitionLimit,
 						},
 					},
 				}); err != nil {

weed/mq/client/sub_client/subscriber.go

@@ -21,7 +21,7 @@ type ContentConfiguration struct {
 }
 
 type ProcessorConfiguration struct {
-	ConcurrentPartitionLimit int // how many partitions to process concurrently
+	ConcurrentPartitionLimit int32 // how many partitions to process concurrently
 }
 
 type OnEachMessageFunc func(key, value []byte) (shouldContinue bool, err error)

weed/mq/sub_coordinator/consumer_group.go

@@ -14,6 +14,7 @@ type ConsumerGroupInstance struct {
 	// the consumer group instance may not have an active partition
 	Partitions   []*topic.Partition
 	ResponseChan chan *mq_pb.SubscriberToSubCoordinatorResponse
+	MaxPartitionCount int32
 }
 type ConsumerGroup struct {
 	topic topic.Topic

weed/mq/sub_coordinator/coordinator.go

@@ -67,6 +67,7 @@ func (c *Coordinator) AddSubscriber(initMessage *mq_pb.SubscriberToSubCoordinato
 			cgi, _ = cg.ConsumerGroupInstances.Get(initMessage.ConsumerGroupInstanceId)
 		}
 	}
+	cgi.MaxPartitionCount = initMessage.MaxPartitionCount
 	cg.OnAddConsumerGroupInstance(initMessage.ConsumerGroupInstanceId, initMessage.Topic)
 	return cgi
 }