Repository: apache/pulsar-client-go (go)

Issue:
Consume Performance drops when set EnableBatchIndexAcknowledgment = true
According to PR: #938 
use master version(v0.9.1-0.20230117072740-d9b18d0690c1) to consume messages while EnableBatchIndexAcknowledgment set  
 true，but **consume performance drops to 2/3 of previous**。

The test situation is as follows:
Topic has 5 partitions, producer production rate is 20MB/s , 300000 rows/s.

and consumers consume situations:
| SDK Version  | Enable Batch Index Ack | Consumer Instances | Consume Rate |
| ------------- | ------------- | ------------- | ------------- |
| v0.9.1-0.20230117072740-d9b18d0690c1 | Yes  | 3 | 100000 rows/s  |
| v0.9.1-0.20230117072740-d9b18d0690c1 | No  | 3 | 300000+ rows/s  |


 Analyze the problem by pprof，we found that internal.(*connection).internalSendRequest and pulsar.(*partitionConsumer).internalAck are much more resource intensive when set EnableBatchIndexAcknowledgment as true.

Review the code:
```
func (pc *partitionConsumer) ackID(msgID MessageID, withResponse bool) error {
	if state := pc.getConsumerState(); state == consumerClosed || state == consumerClosing {
		pc.log.WithField("state", state).Error("Failed to ack by closing or closed consumer")
		return errors.New("consumer state is closed")
	}

	if cmid, ok := toChunkedMessageID(msgID); ok {
		return pc.unAckChunksTracker.ack(cmid)
	}

	trackingID, ok := toTrackingMessageID(msgID)
	if !ok {
		return errors.New("failed to convert trackingMessageID")
	}

	ackReq := new(ackRequest)
	ackReq.doneCh = make(chan struct{})
	ackReq.ackType = individualAck
	if !trackingID.Undefined() && trackingID.ack() {
		pc.metrics.AcksCounter.Inc()
		pc.metrics.ProcessingTime.Observe(float64(time.Now().UnixNano()-trackingID.receivedTime.UnixNano()) / 1.0e9)
		ackReq.msgID = trackingID
		// send ack request to eventsCh
		pc.eventsCh <- ackReq

		if withResponse {
			<-ackReq.doneCh
		}

		pc.options.interceptors.OnAcknowledge(pc.parentConsumer, msgID)
	} else if pc.options.enableBatchIndexAck {
		ackReq.msgID = trackingID
		pc.eventsCh <- ackReq
	}

	if withResponse {
		return ackReq.err
	}
	return nil
}
```

Maybe problem is that partitionConsumer will send ack request to *Pulsar Server* by every MessageID, without waiting all msg of one batch be acked by ackTracker, it leads to ack requests becoming much more than BatchIndexAck disabled, performance drops bacause of much more processing requests. And backlog is lasting increasing, could not catch up with the production rate.


So, enableBatchIndexAck should follow the previous processing method or there is another way.


@BewareMyPower Thank you for developing this feature. Could you take a look at this problem ? Thanks a lot!


Pull request description:
Support grouping ACK requests by time and size
Fixes #949 

### Motivation

Currently the Go client does not support grouping ACK requests, so each time `Ack` (or similar APIs) is called, a ACK request will be sent, which could downgrade the performance. We need to support configuring the time and size to cache `MessageID` before sending ACK requests.

### Modifications
- Add an `AckGroupingOptions` field to `ConsumerOptions`, when it's nil, use 100ms as the max time and 1000 as the max size.
- Add an `ackGroupingTracker` interface to support grouping ACK requests.
- When `AckWithResponse` is false, adding the `MessageID` instance to the tracker instead of sending the requests to `eventsCh`.

### Verifying this change

- [ ] Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:
  - Added `ack_grouping_tracker_test.go` to verify `ackGroupingTracker` in various cases
  - The consumer side change can be covered by existing tests because the default `AckGroupingOptions` config is `{ MaxSize: 1000, MaxTime: 100*time.Millisecond }`.

### Does this pull request potentially affect one of the following parts:

*If `yes` was chosen, please highlight the changes*

  - Dependencies (does it add or upgrade a dependency): (yes / no)
  - The public API: (**yes**)
  - The schema: (yes / no / don't know)
  - The default values of configurations: (yes / no)
  - The wire protocol: (yes / no)

### Documentation

  - Does this pull request introduce a new feature? (yes)
  - If yes, how is the feature documented? (GoDocs)
  - If a feature is not applicable for documentation, explain why?
  - If a feature is not documented yet in this PR, please create a followup issue for adding the documentation


Patch:
diff --git a/pulsar/ack_grouping_tracker.go b/pulsar/ack_grouping_tracker.go
new file mode 100644
index 0000000..dbc70f5
--- /dev/null
+++ b/pulsar/ack_grouping_tracker.go
@@ -0,0 +1,298 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package pulsar
+
+import (
+	"time"
+
+	"github.com/bits-and-blooms/bitset"
+)
+
+type ackGroupingTracker interface {
+	add(id MessageID)
+
+	addCumulative(id MessageID)
+
+	isDuplicate(id MessageID) bool
+
+	flush()
+
+	flushAndClean()
+
+	close()
+}
+
+type ackFlushType int
+
+const (
+	flushOnly ackFlushType = iota
+	flushAndClean
+	flushAndClose
+)
+
+func newAckGroupingTracker(options *AckGroupingOptions,
+	ackIndividual func(id MessageID),
+	ackCumulative func(id MessageID)) ackGroupingTracker {
+	if options == nil {
+		options = &AckGroupingOptions{
+			MaxSize: 1000,
+			MaxTime: 100 * time.Millisecond,
+		}
+	}
+
+	if options.MaxSize <= 1 {
+		return &immediateAckGroupingTracker{
+			ackIndividual: ackIndividual,
+			ackCumulative: ackCumulative,
+		}
+	}
+
+	c := &cachedAcks{
+		singleAcks:        make([]MessageID, options.MaxSize),
+		pendingAcks:       make(map[int64]*bitset.BitSet),
+		lastCumulativeAck: EarliestMessageID(),
+		ackIndividual:     ackIndividual,
+		ackCumulative:     ackCumulative,
+		ackList: func(ids []MessageID) {
+			// TODO: support ack a list of MessageIDs
+			for _, id := range ids {
+				ackIndividual(id)
+			}
+		},
+	}
+
+	timeout := time.NewTicker(time.Hour)
+	if options.MaxTime > 0 {
+		timeout = time.NewTicker(options.MaxTime)
+	} else {
+		timeout.Stop()
+	}
+	t := &timedAckGroupingTracker{
+		ackIndividualCh:   make(chan MessageID),
+		ackCumulativeCh:   make(chan MessageID),
+		duplicateIDCh:     make(chan MessageID),
+		duplicateResultCh: make(chan bool),
+		flushCh:           make(chan ackFlushType),
+		waitFlushCh:       make(chan bool),
+	}
+	go func() {
+		for {
+			select {
+			case id := <-t.ackIndividualCh:
+				if c.addAndCheckIfFull(id) {
+					c.flushIndividualAcks()
+					if options.MaxTime > 0 {
+						timeout.Reset(options.MaxTime)
+					}
+				}
+			case id := <-t.ackCumulativeCh:
+				c.tryUpdateLastCumulativeAck(id)
+				if options.MaxTime <= 0 {
+					c.flushCumulativeAck()
+				}
+			case id := <-t.duplicateIDCh:
+				t.duplicateResultCh <- c.isDuplicate(id)
+			case <-timeout.C:
+				c.flush()
+			case ackFlushType := <-t.flushCh:
+				timeout.Stop()
+				c.flush()
+				if ackFlushType == flushAndClean {
+					c.clean()
+				}
+				t.waitFlushCh <- true
+				if ackFlushType == flushAndClose {
+					return
+				}
+			}
+		}
+	}()
+	return t
+}
+
+type immediateAckGroupingTracker struct {
+	ackIndividual func(id MessageID)
+	ackCumulative func(id MessageID)
+}
+
+func (i *immediateAckGroupingTracker) add(id MessageID) {
+	i.ackIndividual(id)
+}
+
+func (i *immediateAckGroupingTracker) addCumulative(id MessageID) {
+	i.ackCumulative(id)
+}
+
+func (i *immediateAckGroupingTracker) isDuplicate(id MessageID) bool {
+	return false
+}
+
+func (i *immediateAckGroupingTracker) flush() {
+}
+
+func (i *immediateAckGroupingTracker) flushAndClean() {
+}
+
+func (i *immediateAckGroupingTracker) close() {
+}
+
+type cachedAcks struct {
+	singleAcks []MessageID
+	index      int
+
+	// Key is the hash code of the ledger id and the netry id,
+	// Value is the bit set that represents which messages are acknowledged if the entry stores a batch.
+	// The bit 1 represents the message has been acknowledged, i.e. the bits "111" represents all messages
+	// in the batch whose batch size is 3 are not acknowledged.
+	// After the 1st message (i.e. batch index is 0) is acknowledged, the bits will become "011".
+	// Value is nil if the entry represents a single message.
+	pendingAcks map[int64]*bitset.BitSet
+
+	lastCumulativeAck     MessageID
+	cumulativeAckRequired bool
+
+	ackIndividual func(id MessageID)
+	ackCumulative func(id MessageID)
+	ackList       func(ids []MessageID)
+}
+
+func (t *cachedAcks) addAndCheckIfFull(id MessageID) bool {
+	t.singleAcks[t.index] = id
+	t.index++
+	key := messageIDHash(id)
+	ackSet, found := t.pendingAcks[key]
+	if !found {
+		if messageIDIsBatch(id) {
+			ackSet = bitset.New(uint(id.BatchSize()))
+			for i := 0; i < int(id.BatchSize()); i++ {
+				ackSet.Set(uint(i))
+			}
+			t.pendingAcks[key] = ackSet
+		} else {
+			t.pendingAcks[key] = nil
+		}
+	}
+	if ackSet != nil {
+		ackSet.Clear(uint(id.BatchIdx()))
+	}
+	return t.index == len(t.singleAcks)
+}
+
+func (t *cachedAcks) tryUpdateLastCumulativeAck(id MessageID) {
+	if messageIDCompare(t.lastCumulativeAck, id) < 0 {
+		t.lastCumulativeAck = id
+		t.cumulativeAckRequired = true
+	}
+}
+
+func (t *cachedAcks) isDuplicate(id MessageID) bool {
+	if messageIDCompare(t.lastCumulativeAck, id) >= 0 {
+		return true
+	}
+	ackSet, found := t.pendingAcks[messageIDHash(id)]
+	if !found {
+		return false
+	}
+	if ackSet == nil || !messageIDIsBatch(id) {
+		// NOTE: should we panic when ackSet != nil and messageIDIsBatch(id) is true?
+		return true
+	}
+	// 0 represents the message has been acknowledged
+	return !ackSet.Test(uint(id.BatchIdx()))
+}
+
+func (t *cachedAcks) flushIndividualAcks() {
+	if t.index > 0 {
+		t.ackList(t.singleAcks[0:t.index])
+		for _, id := range t.singleAcks[0:t.index] {
+			key := messageIDHash(id)
+			ackSet, found := t.pendingAcks[key]
+			if !found {
+				continue
+			}
+			if ackSet == nil {
+				delete(t.pendingAcks, key)
+			} else {
+				ackSet.Clear(uint(id.BatchIdx()))
+				if ackSet.None() { // all messages have been acknowledged
+					delete(t.pendingAcks, key)
+				}
+			}
+			delete(t.pendingAcks, messageIDHash(id))
+		}
+		t.index = 0
+	}
+}
+
+func (t *cachedAcks) flushCumulativeAck() {
+	if t.cumulativeAckRequired {
+		t.ackCumulative(t.lastCumulativeAck)
+		t.cumulativeAckRequired = false
+	}
+}
+
+func (t *cachedAcks) flush() {
+	t.flushIndividualAcks()
+	t.flushCumulativeAck()
+}
+
+func (t *cachedAcks) clean() {
+	maxSize := len(t.singleAcks)
+	t.singleAcks = make([]MessageID, maxSize)
+	t.index = 0
+	t.pendingAcks = make(map[int64]*bitset.BitSet)
+	t.lastCumulativeAck = EarliestMessageID()
+	t.cumulativeAckRequired = false
+}
+
+type timedAckGroupingTracker struct {
+	ackIndividualCh   chan MessageID
+	ackCumulativeCh   chan MessageID
+	duplicateIDCh     chan MessageID
+	duplicateResultCh chan bool
+	flushCh           chan ackFlushType
+	waitFlushCh       chan bool
+}
+
+func (t *timedAckGroupingTracker) add(id MessageID) {
+	t.ackIndividualCh <- id
+}
+
+func (t *timedAckGroupingTracker) addCumulative(id MessageID) {
+	t.ackCumulativeCh <- id
+}
+
+func (t *timedAckGroupingTracker) isDuplicate(id MessageID) bool {
+	t.duplicateIDCh <- id
+	return <-t.duplicateResultCh
+}
+
+func (t *timedAckGroupingTracker) flush() {
+	t.flushCh <- flushOnly
+	<-t.waitFlushCh
+}
+
+func (t *timedAckGroupingTracker) flushAndClean() {
+	t.flushCh <- flushAndClean
+	<-t.waitFlushCh
+}
+
+func (t *timedAckGroupingTracker) close() {
+	t.flushCh <- flushAndClose
+	<-t.waitFlushCh
+}
diff --git a/pulsar/consumer.go b/pulsar/consumer.go
index 8bae57d..9576d7a 100644
--- a/pulsar/consumer.go
+++ b/pulsar/consumer.go
@@ -80,6 +80,20 @@ type DLQPolicy struct {
 	RetryLetterTopic string
 }
 
+// AckGroupingOptions controls how to group ACK requests
+// If maxSize is 0 or 1, any ACK request will be sent immediately.
+// Otherwise, the ACK requests will be cached until one of the following conditions meets:
+// 1. There are `MaxSize` pending ACK requests.
+// 2. `MaxTime` is greater than 1 microsecond and ACK requests have been cached for `maxTime`.
+// Specially, for cumulative acknowledgment, only the latest ACK is cached and it will only be sent after `MaxTime`.
+type AckGroupingOptions struct {
+	// The maximum number of ACK requests to cache
+	MaxSize uint32
+
+	// The maximum time to cache ACK requests
+	MaxTime time.Duration
+}
+
 // ConsumerOptions is used to configure and create instances of Consumer.
 type ConsumerOptions struct {
 	// Topic specifies the topic this consumer will subscribe on.
@@ -215,6 +229,13 @@ type ConsumerOptions struct {
 	// Enable or disable batch index acknowledgment. To enable this feature, ensure batch index acknowledgment
 	// is enabled on the broker side. (default: false)
 	EnableBatchIndexAcknowledgment bool
+
+	// Controls how to group ACK requests, the default value is nil, which means:
+	// MaxSize: 1000
+	// MaxTime: 100*time.Millisecond
+	// NOTE: This option does not work if AckWithResponse is true
+	//	because there are only synchronous APIs for acknowledgment
+	AckGroupingOptions *AckGroupingOptions
 }
 
 // Consumer is an interface that abstracts behavior of Pulsar's consumer
diff --git a/pulsar/consumer_impl.go b/pulsar/consumer_impl.go
index bf136c8..8ee1822 100644
--- a/pulsar/consumer_impl.go
+++ b/pulsar/consumer_impl.go
@@ -398,6 +398,7 @@ func (c *consumer) internalTopicSubscribeToPartitions() error {
 				autoAckIncompleteChunk:      c.options.AutoAckIncompleteChunk,
 				consumerEventListener:       c.options.EventListener,
 				enableBatchIndexAck:         c.options.EnableBatchIndexAcknowledgment,
+				ackGroupingOptions:          c.options.AckGroupingOptions,
 			}
 			cons, err := newPartitionConsumer(c, c.client, opts, c.messageCh, c.dlq, c.metrics)
 			ch <- ConsumerError{
diff --git a/pulsar/consumer_partition.go b/pulsar/consumer_partition.go
index e723f8a..95a8d32 100644
--- a/pulsar/consumer_partition.go
+++ b/pulsar/consumer_partition.go
@@ -116,6 +116,7 @@ type partitionConsumerOpts struct {
 	// in failover mode, this callback will be called when consumer change
 	consumerEventListener ConsumerEventListener
 	enableBatchIndexAck   bool
+	ackGroupingOptions    *AckGroupingOptions
 }
 
 type ConsumerEventListener interface {
@@ -167,6 +168,7 @@ type partitionConsumer struct {
 
 	chunkedMsgCtxMap   *chunkedMsgCtxMap
 	unAckChunksTracker *unAckChunksTracker
+	ackGroupingTracker ackGroupingTracker
 }
 
 func (pc *partitionConsumer) ActiveConsumerChanged(isActive bool) {
@@ -310,6 +312,9 @@ func newPartitionConsumer(parent Consumer, client *client, options *partitionCon
 	pc.availablePermits = &availablePermits{pc: pc}
 	pc.chunkedMsgCtxMap = newChunkedMsgCtxMap(options.maxPendingChunkedMessage, pc)
 	pc.unAckChunksTracker = newUnAckChunksTracker(pc)
+	pc.ackGroupingTracker = newAckGroupingTracker(options.ackGroupingOptions,
+		func(id MessageID) { pc.sendIndividualAck(id) },
+		func(id MessageID) { pc.sendCumulativeAck(id) })
 	pc.setConsumerState(consumerInit)
 	pc.log = client.log.SubLogger(log.Fields{
 		"name":         pc.name,
@@ -467,30 +472,35 @@ func (pc *partitionConsumer) ackID(msgID MessageID, withResponse bool) error {
 		return errors.New("failed to convert trackingMessageID")
 	}
 
-	ackReq := new(ackRequest)
-	ackReq.doneCh = make(chan struct{})
-	ackReq.ackType = individualAck
 	if !trackingID.Undefined() && trackingID.ack() {
 		pc.metrics.AcksCounter.Inc()
 		pc.metrics.ProcessingTime.Observe(float64(time.Now().UnixNano()-trackingID.receivedTime.UnixNano()) / 1.0e9)
-		ackReq.msgID = trackingID
-		// send ack request to eventsCh
-		pc.eventsCh <- ackReq
-
-		if withResponse {
-			<-ackReq.doneCh
-		}
-
-		pc.options.interceptors.OnAcknowledge(pc.parentConsumer, msgID)
-	} else if pc.options.enableBatchIndexAck {
-		ackReq.msgID = trackingID
-		pc.eventsCh <- ackReq
+	} else if !pc.options.enableBatchIndexAck {
+		return nil
 	}
 
+	var ackReq *ackRequest
 	if withResponse {
-		return ackReq.err
+		ackReq := pc.sendIndividualAck(&trackingID)
+		<-ackReq.doneCh
+	} else {
+		pc.ackGroupingTracker.add(&trackingID)
 	}
-	return nil
+	pc.options.interceptors.OnAcknowledge(pc.parentConsumer, msgID)
+	if ackReq == nil {
+		return nil
+	}
+	return ackReq.err
+}
+
+func (pc *partitionConsumer) sendIndividualAck(msgID MessageID) *ackRequest {
+	ackReq := &ackRequest{
+		doneCh:  make(chan struct{}),
+		ackType: individualAck,
+		msgID:   *msgID.(*trackingMessageID),
+	}
+	pc.eventsCh <- ackReq
+	return ackReq
 }
 
 func (pc *partitionConsumer) AckIDWithResponse(msgID MessageID) error {
@@ -524,14 +534,12 @@ func (pc *partitionConsumer) internalAckIDCumulative(msgID MessageID, withRespon
 		return nil
 	}
 
-	ackReq := new(ackRequest)
-	ackReq.doneCh = make(chan struct{})
-	ackReq.ackType = cumulativeAck
+	var msgIDToAck trackingMessageID
 	if trackingID.ackCumulative() || pc.options.enableBatchIndexAck {
-		ackReq.msgID = trackingID
+		msgIDToAck = trackingID
 	} else if !trackingID.tracker.hasPrevBatchAcked() {
 		// get previous batch message id
-		ackReq.msgID = trackingID.prev()
+		msgIDToAck = trackingID.prev()
 		trackingID.tracker.setPrevBatchAcked()
 	} else {
 		// waiting for all the msgs are acked in this batch
@@ -540,12 +548,13 @@ func (pc *partitionConsumer) internalAckIDCumulative(msgID MessageID, withRespon
 
 	pc.metrics.AcksCounter.Inc()
 	pc.metrics.ProcessingTime.Observe(float64(time.Now().UnixNano()-trackingID.receivedTime.UnixNano()) / 1.0e9)
-	// send ack request to eventsCh
-	pc.eventsCh <- ackReq
 
+	var ackReq *ackRequest
 	if withResponse {
-		// wait for the request to complete if withResponse set true
+		ackReq := pc.sendCumulativeAck(&msgIDToAck)
 		<-ackReq.doneCh
+	} else {
+		pc.ackGroupingTracker.addCumulative(&msgIDToAck)
 	}
 
 	pc.options.interceptors.OnAcknowledge(pc.parentConsumer, msgID)
@@ -554,7 +563,20 @@ func (pc *partitionConsumer) internalAckIDCumulative(msgID MessageID, withRespon
 		pc.unAckChunksTracker.remove(cmid)
 	}
 
-	return nil
+	if ackReq == nil {
+		return nil
+	}
+	return ackReq.err
+}
+
+func (pc *partitionConsumer) sendCumulativeAck(msgID MessageID) *ackRequest {
+	ackReq := &ackRequest{
+		doneCh:  make(chan struct{}),
+		ackType: cumulativeAck,
+		msgID:   *msgID.(*trackingMessageID),
+	}
+	pc.eventsCh <- ackReq
+	return ackReq
 }
 
 func (pc *partitionConsumer) NackID(msgID MessageID) {
@@ -631,6 +653,9 @@ func (pc *partitionConsumer) Close() {
 		return
 	}
 
+	// flush all pending ACK requests and terminate the timer goroutine
+	pc.ackGroupingTracker.close()
+
 	// close chunkedMsgCtxMap
 	pc.chunkedMsgCtxMap.Close()
 
@@ -658,6 +683,7 @@ func (pc *partitionConsumer) Seek(msgID MessageID) error {
 		return errors.New("unhandled messageID type")
 	}
 
+	pc.ackGroupingTracker.flushAndClean()
 	pc.eventsCh <- req
 
 	// wait for the request to complete
@@ -715,6 +741,7 @@ func (pc *partitionConsumer) SeekByTime(time time.Time) error {
 		doneCh:      make(chan struct{}),
 		publishTime: time,
 	}
+	pc.ackGroupingTracker.flushAndClean()
 	pc.eventsCh <- req
 
 	// wait for the request to complete
@@ -957,6 +984,10 @@ func (pc *partitionConsumer) MessageReceived(response *pb.CommandMessage, header
 			msgID = trackingMsgID
 		}
 
+		if pc.ackGroupingTracker.isDuplicate(msgID) {
+			continue
+		}
+
 		var messageIndex *uint64
 		var brokerPublishTime *time.Time
 		if brokerMetadata != nil {
diff --git a/pulsar/internal/connection.go b/pulsar/internal/connection.go
index cb4af33..67b6f32 100644
--- a/pulsar/internal/connection.go
+++ b/pulsar/internal/connection.go
@@ -523,8 +523,14 @@ func (c *connection) internalReceivedCommand(cmd *pb.BaseCommand, headersAndPayl
 		c.handleResponse(cmd.Success.GetRequestId(), cmd)
 
 	case pb.BaseCommand_PRODUCER_SUCCESS:
-		c.handleResponse(cmd.ProducerSuccess.GetRequestId(), cmd)
-
+		if !*cmd.ProducerSuccess.ProducerReady {
+			request, ok := c.findPendingRequest(cmd.ProducerSuccess.GetRequestId())
+			if ok {
+				request.callback(cmd, nil)
+			}
+		} else {
+			c.handleResponse(cmd.ProducerSuccess.GetRequestId(), cmd)
+		}
 	case pb.BaseCommand_PARTITIONED_METADATA_RESPONSE:
 		c.checkServerError(cmd.PartitionMetadataResponse.Error)
 		c.handleResponse(cmd.PartitionMetadataResponse.GetRequestId(), cmd)
@@ -748,6 +754,13 @@ func (c *connection) deletePendingRequest(requestID uint64) (*request, bool) {
 	return request, ok
 }
 
+func (c *connection) findPendingRequest(requestID uint64) (*request, bool) {
+	c.pendingLock.Lock()
+	defer c.pendingLock.Unlock()
+	request, ok := c.pendingReqs[requestID]
+	return request, ok
+}
+
 func (c *connection) failPendingRequests(err error) bool {
 	c.pendingLock.Lock()
 	defer c.pendingLock.Unlock()
diff --git a/pulsar/internal/rpc_client.go b/pulsar/internal/rpc_client.go
index 378ab4f..0ee8ca9 100644
--- a/pulsar/internal/rpc_client.go
+++ b/pulsar/internal/rpc_client.go
@@ -129,14 +129,24 @@ func (c *rpcClient) Request(logicalAddr *url.URL, physicalAddr *url.URL, request
 			Cnx:      cnx,
 			Response: response,
 		}, err}
-		close(ch)
 	})
 
-	select {
-	case res := <-ch:
-		return res.RPCResult, res.error
-	case <-time.After(c.requestTimeout):
-		return nil, ErrRequestTimeOut
+	timeoutCh := time.After(c.requestTimeout)
+	for {
+		select {
+		case res := <-ch:
+			// Ignoring producer not ready response.
+			// Continue to wait for the producer to create successfully
+			if res.error == nil && *res.RPCResult.Response.Type == pb.BaseCommand_PRODUCER_SUCCESS {
+				if !*res.RPCResult.Response.ProducerSuccess.ProducerReady {
+					timeoutCh = nil
+					break
+				}
+			}
+			return res.RPCResult, res.error
+		case <-timeoutCh:
+			return nil, ErrRequestTimeOut
+		}
 	}
 }
 
diff --git a/pulsar/message.go b/pulsar/message.go
index d37692b..c44957d 100644
--- a/pulsar/message.go
+++ b/pulsar/message.go
@@ -181,3 +181,42 @@ func EarliestMessageID() MessageID {
 func LatestMessageID() MessageID {
 	return latestMessageID
 }
+
+func messageIDCompare(lhs MessageID, rhs MessageID) int {
+	if lhs.LedgerID() < rhs.LedgerID() {
+		return -1
+	} else if lhs.LedgerID() > rhs.LedgerID() {
+		return 1
+	}
+	if lhs.EntryID() < rhs.EntryID() {
+		return -1
+	} else if lhs.EntryID() > rhs.EntryID() {
+		return 1
+	}
+	// When performing batch index ACK on a batched message whose batch size is N,
+	// the ACK order should be:
+	//   (ledger, entry, 0) -> (ledger, entry, 1) -> ... -> (ledger, entry, N-1) -> (ledger, entry)
+	// So we have to treat any MessageID with the batch index precedes the MessageID without the batch index
+	// if they are in the same entry.
+	if lhs.BatchIdx() < 0 && rhs.BatchIdx() < 0 {
+		return 0
+	} else if lhs.BatchIdx() >= 0 && rhs.BatchIdx() < 0 {
+		return -1
+	} else if lhs.BatchIdx() < 0 && rhs.BatchIdx() >= 0 {
+		return 1
+	}
+	if lhs.BatchIdx() < rhs.BatchIdx() {
+		return -1
+	} else if lhs.BatchIdx() > rhs.BatchIdx() {
+		return 1
+	}
+	return 0
+}
+
+func messageIDHash(id MessageID) int64 {
+	return id.LedgerID() + 31*id.EntryID()
+}
+
+func messageIDIsBatch(id MessageID) bool {
+	return id.BatchIdx() >= 0 && id.BatchSize() > 0
+}
diff --git a/pulsar/producer.go b/pulsar/producer.go
index d9ac34b..8fcb891 100644
--- a/pulsar/producer.go
+++ b/pulsar/producer.go
@@ -64,6 +64,9 @@ const (
 	// ProducerAccessModeExclusive is required exclusive access for producer.
 	// Fail immediately if there's already a producer connected.
 	ProducerAccessModeExclusive
+
+	// ProducerAccessModeWaitForExclusive is pending until producer can acquire exclusive access.
+	ProducerAccessModeWaitForExclusive
 )
 
 // TopicMetadata represents a topic metadata.
diff --git a/pulsar/producer_partition.go b/pulsar/producer_partition.go
index eece055..160693c 100644
--- a/pulsar/producer_partition.go
+++ b/pulsar/producer_partition.go
@@ -1367,6 +1367,8 @@ func toProtoProducerAccessMode(accessMode ProducerAccessMode) pb.ProducerAccessM
 		return pb.ProducerAccessMode_Shared
 	case ProducerAccessModeExclusive:
 		return pb.ProducerAccessMode_Exclusive
+	case ProducerAccessModeWaitForExclusive:
+		return pb.ProducerAccessMode_WaitForExclusive
 	}
 
 	return pb.ProducerAccessMode_Shared


Test patch:
diff --git a/pulsar/ack_grouping_tracker_test.go b/pulsar/ack_grouping_tracker_test.go
new file mode 100644
index 0000000..d7903e8
--- /dev/null
+++ b/pulsar/ack_grouping_tracker_test.go
@@ -0,0 +1,197 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package pulsar
+
+import (
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestNoCacheTracker(t *testing.T) {
+	tests := []AckGroupingOptions{
+		{
+			MaxSize: 0,
+			MaxTime: 10 * time.Hour,
+		},
+		{
+			MaxSize: 1,
+			MaxTime: 10 * time.Hour,
+		},
+	}
+	for _, option := range tests {
+		t.Run(fmt.Sprintf("TestAckImmediately_size_%v_time_%vs", option.MaxSize, option.MaxTime.Seconds()),
+			func(t *testing.T) {
+				ledgerID0 := int64(-1)
+				ledgerID1 := int64(-1)
+				tracker := newAckGroupingTracker(&option,
+					func(id MessageID) { ledgerID0 = id.LedgerID() },
+					func(id MessageID) { ledgerID1 = id.LedgerID() })
+
+				tracker.add(&messageID{ledgerID: 1})
+				assert.Equal(t, atomic.LoadInt64(&ledgerID0), int64(1))
+				tracker.addCumulative(&messageID{ledgerID: 2})
+				assert.Equal(t, atomic.LoadInt64(&ledgerID1), int64(2))
+			})
+	}
+}
+
+type mockAcker struct {
+	sync.Mutex
+	ledgerIDs          []int64
+	cumulativeLedgerID int64
+}
+
+func (a *mockAcker) ack(id MessageID) {
+	defer a.Unlock()
+	a.Lock()
+	a.ledgerIDs = append(a.ledgerIDs, id.LedgerID())
+}
+
+func (a *mockAcker) ackCumulative(id MessageID) {
+	atomic.StoreInt64(&a.cumulativeLedgerID, id.LedgerID())
+}
+
+func (a *mockAcker) getLedgerIDs() []int64 {
+	defer a.Unlock()
+	a.Lock()
+	return a.ledgerIDs
+}
+
+func (a *mockAcker) getCumulativeLedgerID() int64 {
+	return atomic.LoadInt64(&a.cumulativeLedgerID)
+}
+
+func (a *mockAcker) reset() {
+	a.ledgerIDs = make([]int64, 0)
+	a.cumulativeLedgerID = int64(0)
+}
+
+func TestCachedTracker(t *testing.T) {
+	var acker mockAcker
+	tracker := newAckGroupingTracker(&AckGroupingOptions{MaxSize: 3, MaxTime: 0},
+		func(id MessageID) { acker.ack(id) }, func(id MessageID) { acker.ackCumulative(id) })
+
+	tracker.add(&messageID{ledgerID: 1})
+	tracker.add(&messageID{ledgerID: 2})
+	for i := 1; i <= 2; i++ {
+		assert.True(t, tracker.isDuplicate(&messageID{ledgerID: int64(i)}))
+	}
+	assert.Equal(t, 0, len(acker.getLedgerIDs()))
+	tracker.add(&messageID{ledgerID: 3})
+	assert.Eventually(t, func() bool { return len(acker.getLedgerIDs()) > 0 },
+		10*time.Millisecond, 2*time.Millisecond)
+	assert.Equal(t, []int64{1, 2, 3}, acker.getLedgerIDs())
+	for i := 1; i <= 3; i++ {
+		assert.False(t, tracker.isDuplicate(&messageID{ledgerID: int64(i)}))
+	}
+
+	tracker.add(&messageID{ledgerID: 4})
+	// 4 won't be added because the cache is not full
+	assert.Equal(t, []int64{1, 2, 3}, acker.getLedgerIDs())
+
+	assert.False(t, tracker.isDuplicate(&messageID{ledgerID: 5}))
+	tracker.addCumulative(&messageID{ledgerID: 5})
+	for i := 0; i <= 5; i++ {
+		assert.True(t, tracker.isDuplicate(&messageID{ledgerID: int64(i)}))
+	}
+	assert.Equal(t, int64(5), acker.getCumulativeLedgerID())
+	assert.False(t, tracker.isDuplicate(&messageID{ledgerID: int64(6)}))
+
+	tracker.flush()
+	assert.Eventually(t, func() bool { return len(acker.getLedgerIDs()) > 3 },
+		10*time.Millisecond, 2*time.Millisecond)
+	assert.Equal(t, []int64{1, 2, 3, 4}, acker.getLedgerIDs())
+}
+
+func TestTimedTrackerIndividualAck(t *testing.T) {
+	var acker mockAcker
+	// MaxSize: 1000, MaxTime: 100ms
+	tracker := newAckGroupingTracker(nil, func(id MessageID) { acker.ack(id) }, nil)
+
+	expected := make([]int64, 0)
+	for i := 0; i < 999; i++ {
+		tracker.add(&messageID{ledgerID: int64(i)})
+		expected = append(expected, int64(i))
+	}
+	assert.Equal(t, 0, len(acker.getLedgerIDs()))
+
+	// case 1: flush because the tracker timed out
+	assert.Eventually(t, func() bool { return len(acker.getLedgerIDs()) == 999 },
+		150*time.Millisecond, 10*time.Millisecond)
+	assert.Equal(t, expected, acker.getLedgerIDs())
+
+	// case 2: flush because cache is full
+	time.Sleep(50) // see case 3
+	acker.reset()
+	expected = append(expected, 999)
+	for i := 0; i < 1001; i++ {
+		tracker.add(&messageID{ledgerID: int64(i)})
+	}
+	assert.Equal(t, expected, acker.getLedgerIDs())
+
+	// case 3: flush will reset the timer
+	start := time.Now()
+	assert.Eventually(t, func() bool { return len(acker.getLedgerIDs()) > 1000 },
+		150*time.Millisecond, 10*time.Millisecond)
+	elapsed := time.Since(start)
+	assert.GreaterOrEqual(t, elapsed, int64(100), "elapsed", elapsed)
+	assert.Equal(t, append(expected, 1000), acker.getLedgerIDs())
+}
+
+func TestTimedTrackerCumulativeAck(t *testing.T) {
+	var acker mockAcker
+	// MaxTime is 100ms
+	tracker := newAckGroupingTracker(nil, nil, func(id MessageID) { acker.ackCumulative(id) })
+
+	// case 1: flush because of the timeout
+	tracker.addCumulative(&messageID{ledgerID: 1})
+	assert.NotEqual(t, int64(1), acker.getCumulativeLedgerID())
+	assert.Eventually(t, func() bool { return acker.getCumulativeLedgerID() == int64(1) },
+		150*time.Millisecond, 10*time.Millisecond)
+	assert.Equal(t, int64(1), acker.getCumulativeLedgerID())
+
+	// case 2: flush manually
+	tracker.addCumulative(&messageID{ledgerID: 2})
+	tracker.flush()
+	assert.Equal(t, int64(2), acker.getCumulativeLedgerID())
+
+	// case 3: older MessageID cannot be acknowledged
+	tracker.addCumulative(&messageID{ledgerID: 1})
+	tracker.flush()
+	assert.Equal(t, int64(2), acker.getCumulativeLedgerID())
+}
+
+func TestTimedTrackerIsDuplicate(t *testing.T) {
+	tracker := newAckGroupingTracker(nil, func(id MessageID) {}, func(id MessageID) {})
+
+	tracker.add(messageID{batchIdx: 0, batchSize: 3})
+	tracker.add(messageID{batchIdx: 2, batchSize: 3})
+	assert.True(t, tracker.isDuplicate(messageID{batchIdx: 0, batchSize: 3}))
+	assert.False(t, tracker.isDuplicate(messageID{batchIdx: 1, batchSize: 3}))
+	assert.True(t, tracker.isDuplicate(messageID{batchIdx: 2, batchSize: 3}))
+
+	tracker.flush()
+	assert.False(t, tracker.isDuplicate(messageID{batchIdx: 0, batchSize: 3}))
+	assert.False(t, tracker.isDuplicate(messageID{batchIdx: 1, batchSize: 3}))
+	assert.False(t, tracker.isDuplicate(messageID{batchIdx: 2, batchSize: 3}))
+}
diff --git a/pulsar/consumer_partition_test.go b/pulsar/consumer_partition_test.go
index fd50b70..b9a9a02 100644
--- a/pulsar/consumer_partition_test.go
+++ b/pulsar/consumer_partition_test.go
@@ -37,6 +37,8 @@ func TestSingleMessageIDNoAckTracker(t *testing.T) {
 		metrics:              newTestMetrics(),
 		decryptor:            crypto.NewNoopDecryptor(),
 	}
+	pc.ackGroupingTracker = newAckGroupingTracker(&AckGroupingOptions{MaxSize: 0},
+		func(id MessageID) { pc.sendIndividualAck(id) }, nil)
 
 	headersAndPayload := internal.NewBufferWrapper(rawCompatSingleMessage)
 	if err := pc.MessageReceived(nil, headersAndPayload); err != nil {
@@ -73,6 +75,8 @@ func TestBatchMessageIDNoAckTracker(t *testing.T) {
 		metrics:              newTestMetrics(),
 		decryptor:            crypto.NewNoopDecryptor(),
 	}
+	pc.ackGroupingTracker = newAckGroupingTracker(&AckGroupingOptions{MaxSize: 0},
+		func(id MessageID) { pc.sendIndividualAck(id) }, nil)
 
 	headersAndPayload := internal.NewBufferWrapper(rawBatchMessage1)
 	if err := pc.MessageReceived(nil, headersAndPayload); err != nil {
@@ -105,6 +109,8 @@ func TestBatchMessageIDWithAckTracker(t *testing.T) {
 		metrics:              newTestMetrics(),
 		decryptor:            crypto.NewNoopDecryptor(),
 	}
+	pc.ackGroupingTracker = newAckGroupingTracker(&AckGroupingOptions{MaxSize: 0},
+		func(id MessageID) { pc.sendIndividualAck(id) }, nil)
 
 	headersAndPayload := internal.NewBufferWrapper(rawBatchMessage10)
 	if err := pc.MessageReceived(nil, headersAndPayload); err != nil {
diff --git a/pulsar/consumer_test.go b/pulsar/consumer_test.go
index 0eb7aae..de90c0e 100644
--- a/pulsar/consumer_test.go
+++ b/pulsar/consumer_test.go
@@ -3853,37 +3853,38 @@ func TestAckWithMessageID(t *testing.T) {
 }
 
 func TestBatchIndexAck(t *testing.T) {
-	tests := []struct {
-		AckWithResponse bool
-		Cumulative      bool
-	}{
-		{
-			AckWithResponse: true,
-			Cumulative:      true,
-		},
-		{
-			AckWithResponse: true,
-			Cumulative:      false,
-		},
-		{
-			AckWithResponse: false,
-			Cumulative:      true,
-		},
-		{
-			AckWithResponse: false,
-			Cumulative:      false,
-		},
-	}
-	for _, params := range tests {
-		t.Run(fmt.Sprintf("TestBatchIndexAck_WithResponse_%v_Cumulative_%v",
-			params.AckWithResponse, params.Cumulative),
+	type config struct {
+		ackWithResponse    bool
+		cumulative         bool
+		ackGroupingOptions *AckGroupingOptions
+	}
+	configs := make([]config, 0)
+	for _, option := range []*AckGroupingOptions{
+		nil, // MaxSize: 1000, MaxTime: 10ms
+		{MaxSize: 0, MaxTime: 0},
+		{MaxSize: 1000, MaxTime: 0},
+	} {
+		configs = append(configs, config{true, true, option})
+		configs = append(configs, config{true, false, option})
+		configs = append(configs, config{false, true, option})
+		configs = append(configs, config{false, false, option})
+	}
+
+	for _, params := range configs {
+		option := params.ackGroupingOptions
+		if option == nil {
+			option = &AckGroupingOptions{1000, 10 * time.Millisecond}
+		}
+
+		t.Run(fmt.Sprintf("TestBatchIndexAck_WithResponse_%v_Cumulative_%v_AckGroupingOption_%v_%v",
+			params.ackWithResponse, params.cumulative, option.MaxSize, option.MaxTime.Milliseconds()),
 			func(t *testing.T) {
-				runBatchIndexAckTest(t, params.AckWithResponse, params.Cumulative)
+				runBatchIndexAckTest(t, params.ackWithResponse, params.cumulative, params.ackGroupingOptions)
 			})
 	}
 }
 
-func runBatchIndexAckTest(t *testing.T, ackWithResponse bool, cumulative bool) {
+func runBatchIndexAckTest(t *testing.T, ackWithResponse bool, cumulative bool, option *AckGroupingOptions) {
 	client, err := NewClient(ClientOptions{
 		URL: lookupURL,
 	})
@@ -3897,6 +3898,7 @@ func runBatchIndexAckTest(t *testing.T, ackWithResponse bool, cumulative bool) {
 			SubscriptionName:               "my-sub",
 			AckWithResponse:                ackWithResponse,
 			EnableBatchIndexAcknowledgment: true,
+			AckGroupingOptions:             option,
 		})
 		assert.Nil(t, err)
 		return consumer
diff --git a/pulsar/producer_test.go b/pulsar/producer_test.go
index d7950eb..e69a14c 100644
--- a/pulsar/producer_test.go
+++ b/pulsar/producer_test.go
@@ -1737,3 +1737,45 @@ func TestExclusiveProducer(t *testing.T) {
 	assert.Error(t, err, "Producer should be failed")
 	assert.True(t, strings.Contains(err.Error(), "ProducerBusy"))
 }
+
+func TestWaitForExclusiveProducer(t *testing.T) {
+	client, err := NewClient(ClientOptions{
+		URL: serviceURL,
+		// Set the request timeout is 200ms
+		OperationTimeout: 200 * time.Millisecond,
+	})
+	assert.NoError(t, err)
+	defer client.Close()
+
+	topicName := newTopicName()
+	producer1, err := client.CreateProducer(ProducerOptions{
+		Topic:              topicName,
+		ProducerAccessMode: ProducerAccessModeExclusive,
+	})
+	assert.NoError(t, err)
+	assert.NotNil(t, producer1)
+
+	wg := sync.WaitGroup{}
+	wg.Add(1)
+	go func() {
+		producer2, err := client.CreateProducer(ProducerOptions{
+			Topic:              topicName,
+			ProducerAccessMode: ProducerAccessModeWaitForExclusive,
+		})
+		defer producer2.Close()
+		assert.NoError(t, err)
+		assert.NotNil(t, producer2)
+
+		id, err := producer2.Send(context.Background(), &ProducerMessage{
+			Payload: make([]byte, 1024),
+		})
+		assert.Nil(t, err)
+		assert.NotNil(t, id)
+		wg.Done()
+	}()
+	// Because set the request timeout is 200ms before.
+	// Here waite 300ms to cover wait for exclusive producer never timeout
+	time.Sleep(300 * time.Millisecond)
+	producer1.Close()
+	wg.Wait()
+}