[GitHub] [pulsar-client-go] shibd commented on a diff in pull request #976: [feat]: Support auto scaled consumer receiver queue

["shibd (via GitHub)" <gi...@apache.org>]

shibd commented on code in PR #976:
URL: https://github.com/apache/pulsar-client-go/pull/976#discussion_r1125680331


##########
pulsar/consumer_partition.go:
##########
@@ -1187,35 +1220,48 @@ func (pc *partitionConsumer) dispatcher() {
 	defer func() {
 		pc.log.Debug("exiting dispatch loop")
 	}()
-	var messages []*message
-	for {
-		var queueCh chan []*message
-		var messageCh chan ConsumerMessage
-		var nextMessage ConsumerMessage
 
-		// are there more messages to send?
-		if len(messages) > 0 {
-			nextMessage = ConsumerMessage{
-				Consumer: pc.parentConsumer,
-				Message:  messages[0],
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go func() {
+		var messageCh chan ConsumerMessage
+		for {
+			if len(pc.queueCh) == 0 {

Review Comment:
   Why need a new goroutine to handle this logic?  Can we directly determine whether we need to scale after push nextMessage to `messageCh`?
   
   https://github.com/apache/pulsar-client-go/blob/42ded0d59c46fd3fdaad45f045f7e8bf091131a5/pulsar/consumer_partition.go#L1249-L1255



##########
pulsar/consumer_partition.go:
##########
@@ -1226,34 +1272,16 @@ func (pc *partitionConsumer) dispatcher() {
 			}
 			pc.log.Debug("dispatcher received connection event")
 
-			messages = nil
-
 			// reset available permits
 			pc.availablePermits.reset()
-			initialPermits := uint32(pc.queueSize)
+			initialPermits := uint32(pc.maxQueueSize)

Review Comment:
   We need set `initialReceiverQueueSize `?



##########
pulsar/consumer_partition.go:
##########
@@ -1045,8 +1072,14 @@ func (pc *partitionConsumer) MessageReceived(response *pb.CommandMessage, header
 		messages = append(messages, msg)
 	}
 
+	if len(pc.queueCh) == 0 && len(pc.messageCh) == 0 {

Review Comment:
   Why judge the length of `queueCh` and `messageCh`?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org