track offset

5 files changed, 9 insertions, 6 deletions

diff --git a/weed/mq/client/cmd/weed_sub_kv/subscriber_kv.go b/weed/mq/client/cmd/weed_sub_kv/subscriber_kv.go
index 5286c229d..4bbb26032 100644
--- a/weed/mq/client/cmd/weed_sub_kv/subscriber_kv.go
+++ b/weed/mq/client/cmd/weed_sub_kv/subscriber_kv.go
@@ -37,7 +37,7 @@ func main() {
 	}
 
 	processorConfig := sub_client.ProcessorConfiguration{
-		ConcurrentPartitionLimit: 3,
+		MaxPartitionCount: 3,
 	}
 
 	brokers := strings.Split(*seedBrokers, ",")
diff --git a/weed/mq/client/cmd/weed_sub_record/subscriber_record.go b/weed/mq/client/cmd/weed_sub_record/subscriber_record.go
index a5f87a3bb..ed710fa57 100644
--- a/weed/mq/client/cmd/weed_sub_record/subscriber_record.go
+++ b/weed/mq/client/cmd/weed_sub_record/subscriber_record.go
@@ -63,7 +63,8 @@ func main() {
 	}
 
 	processorConfig := sub_client.ProcessorConfiguration{
-		ConcurrentPartitionLimit: 3,
+		MaxPartitionCount: 3,
+		PerPartitionConcurrency: 1,
 	}
 
 	brokers := strings.Split(*seedBrokers, ",")
diff --git a/weed/mq/client/sub_client/connect_to_sub_coordinator.go b/weed/mq/client/sub_client/connect_to_sub_coordinator.go
index 094ce46ef..815694a48 100644
--- a/weed/mq/client/sub_client/connect_to_sub_coordinator.go
+++ b/weed/mq/client/sub_client/connect_to_sub_coordinator.go
@@ -51,7 +51,7 @@ func (sub *TopicSubscriber) doKeepConnectedToSubCoordinator() {
 							ConsumerGroup:           sub.SubscriberConfig.ConsumerGroup,
 							ConsumerGroupInstanceId: sub.SubscriberConfig.ConsumerGroupInstanceId,
 							Topic:                   sub.ContentConfig.Topic.ToPbTopic(),
-							MaxPartitionCount: 	 	 sub.ProcessorConfig.ConcurrentPartitionLimit,
+							MaxPartitionCount: 	 	 sub.ProcessorConfig.MaxPartitionCount,
 						},
 					},
 				}); err != nil {
@@ -107,6 +107,7 @@ func (sub *TopicSubscriber) onEachPartition(assigned *mq_pb.BrokerPartitionAssig
 					},
 					Filter: sub.ContentConfig.Filter,
 					FollowerBroker: assigned.FollowerBroker,
+					Concurrency: sub.ProcessorConfig.PerPartitionConcurrency,
 				},
 			},
 		});err != nil {
@@ -124,7 +125,7 @@ func (sub *TopicSubscriber) onEachPartition(assigned *mq_pb.BrokerPartitionAssig
 			close(partitionOffsetChan)
 		}()
 
-		concurrentPartitionLimit := int(sub.ProcessorConfig.ConcurrentPartitionLimit)
+		concurrentPartitionLimit := int(sub.ProcessorConfig.MaxPartitionCount)
 		if concurrentPartitionLimit <= 0 {
 			concurrentPartitionLimit = 1
 		}
diff --git a/weed/mq/client/sub_client/subscribe.go b/weed/mq/client/sub_client/subscribe.go
index 950d2214c..ba20cf040 100644
--- a/weed/mq/client/sub_client/subscribe.go
+++ b/weed/mq/client/sub_client/subscribe.go
@@ -29,7 +29,7 @@ func (sub *TopicSubscriber) startProcessors() {
 	// listen to the messages from the sub coordinator
 	// start one processor per partition
 	var wg sync.WaitGroup
-	semaphore := make(chan struct{}, sub.ProcessorConfig.ConcurrentPartitionLimit)
+	semaphore := make(chan struct{}, sub.ProcessorConfig.MaxPartitionCount)
 
 	for assigned := range sub.brokerPartitionAssignmentChan {
 		wg.Add(1)
diff --git a/weed/mq/client/sub_client/subscriber.go b/weed/mq/client/sub_client/subscriber.go
index c7fa26739..95320b19a 100644
--- a/weed/mq/client/sub_client/subscriber.go
+++ b/weed/mq/client/sub_client/subscriber.go
@@ -22,7 +22,8 @@ type ContentConfiguration struct {
 }
 
 type ProcessorConfiguration struct {
-	ConcurrentPartitionLimit int32 // how many partitions to process concurrently
+	MaxPartitionCount int32 // how many partitions to process concurrently
+	PerPartitionConcurrency int32 // how many messages to process concurrently per partition
 }
 
 type OnEachMessageFunc func(key, value []byte) (err error)