Repository: apache/pulsar-client-go
Pull request #1301 (2024-10-29 08:49:46)

Support acknowledging a list of message IDs
### Motivation

For `Shared` and `Key_Shared` subscriptions, if some messages failed to acknowledge, these messages would never be dispatched to the consumer until restarted. If the number of unacknowledged messages reached the threshold, the broker would never dispatch messages anymore. However, the consumer does not have a chance to check which messages failed to acknowledged.

Even if this case was not hit, if the consumer restarted after consuming many messages, the old unacknowledged messages would be delivered again, which is very confusing and might affect the business logic.

Therefore, we can only enable `AckWithResponse` to know which messages failed to acknowledge. Unfortunately, currently the Go SDK only supports acknowledging single messages. It harms the performance significantly.

To solve this solution, this PR adds an API to acknowledge a list of messages.

```golang
	// AckIDList the consumption of a list of messages, identified by their MessageIDs
	// Returns a map of MessageID to error, the keys are the MessageIDs that failed to be acknowledged
	AckIDList([]MessageID) map[MessageID]error
```

Users can save the failed message IDs and add them again in the next `AckIDList` call.

### Modifications

- Add an `AckIDList` API and reuse the logic of `ack_grouping_tracker.go` to convert user provided message IDs to the message IDs in the ACK requests
- Add the request id to the ACK request in `internalAckList` and wait for the response error via a error channel
- Add `TestAckIDList` to verify the case that a message ID list has message IDs of non-batched messages, whole batched messages or partial batched messages because the behaviors are different if the batch index ACK is enabled
- Add `TestMultiTopicAckIDList` to verify the multi-topics case, including regex subscription.

### 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? (not applicable / docs / GoDocs / not documented)
  - 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


nodece: This feature is good to me, I have some questions:

1. Why does the `AckIDList` not support ack the incomplete batch message, when the users use that, it is confusing. 
2. Why not refactor the ` ackIDCommon(msgID MessageID, withResponse bool, txn Transaction)`, we can use `trackingMessageID` to track the batch message. Right now , the `ackIDCommon` and `AckIDList` have the duplication code.
BewareMyPower: ### The 1st question

What's confusing is the tracking message ID itself. The existing `AckID` method's semantic is wrong. Let's look at the following example with `AckID`:

```golang
func TestMyAck(t *testing.T) {
	client, err := NewClient(ClientOptions{
		URL: lookupURL,
	})
	assert.Nil(t, err)
	defer client.Close()

	topic := fmt.Sprintf("test-my-ack-%v", time.Now().Nanosecond())
	createConsumer := func() Consumer {
		consumer, err := client.Subscribe(ConsumerOptions{
			Topic:                       topic,
			SubscriptionName:            "my-sub",
			SubscriptionInitialPosition: SubscriptionPositionEarliest,
			Type:                        Shared,
			AckWithResponse:             true,
		})
		assert.Nil(t, err)
		return consumer
	}
	consumer := createConsumer()
	sendMessages(t, client, topic, 0, 2, true) // send 0 and 1 in the same batch
	msgs := receiveMessages(t, consumer, 2)
	for i := 0; i < 2; i++ {
		fmt.Println("Received message: ", string(msgs[i].Payload()), msgs[i].ID())
	}

	if err := consumer.AckID(msgs[0].ID()); err != nil {
		fmt.Println("Ack message 0 failed: ", err.Error())
	} else {
		fmt.Println("Ack message 0 success")
	}
	consumer.Close()

	consumer = createConsumer()
	msgs = receiveMessages(t, consumer, 1)
	fmt.Println("Received message: ", string(msgs[0].Payload()), msgs[0].ID())
}
```

Outputs:

```
Received message:  msg-0 9:0:0
Received message:  msg-1 9:0:0
Ack message 0 success
Received message:  msg-0 9:0:0
```

From the perspective from user side:
- "msg-0"'s message ID has been acknowledged successfully
- "msg-0" was delivered again after restarting the consumer

Actually this API implements `AckWithResponse` with correct semantics. Replacing the `AckID` call with:

```golang
	for msgID, err := range consumer.AckIDList([]MessageID{msgs[0].ID()}) {
		fmt.Println("Failed to acknowledge ", msgID, err.Error())
	}
```

The outputs will be:

```
Received message:  msg-0 10:0:0
Received message:  msg-1 10:0:0
Failed to acknowledge  10:0:0 incomplete batch
Received message:  msg-0 10:0:0
```

P.S. we should fix the `String()` method of message ID.

Users should add the failed message ID to the next message ID list passed to `AckIDList`. Acknowledging the same message ID 

IMO, we should make batch index ACK enabled by default for both client side and server side. The current default behavior is really confusing.

### The 2nd question

`ackIDCommon` is an abstraction for acknowledgment on a single message ID. When `AckWithResponse` is true, we cannot reuse this method because we should avoid sending N requests.
nodece: @BewareMyPower 

1. The 1st answer

Your explanation is clear, I agree with you:

> IMO, we should make batch index ACK enabled by default for both client side and server side. The current default behavior is really confusing.

Default to batch index ACK is disabled. When disabled, the users must ack all batch messages by `AckID(id)`,  and then the client sends the ack request to the broker. 

> Received message:  msg-0 9:0:0
> Received message:  msg-1 9:0:0
> Ack message 0 success
> Received message:  msg-0 9:0:0

You only ack the first message, the client doesn't send the ack request to the broker, so you still receive the first message after restart the consumer.

I remember the Java client has the same behavior, if I am wrong, please let me know.

2. The 2nd answer.

> `ackIDCommon` is an abstraction for acknowledgment on a single message ID. When `AckWithResponse` is true, we cannot reuse this method because we should avoid sending N requests.

You can refactor the `ackIDCommon`, because you ignore the ackTracer, so it is confusing. I hope we still use the ackTracer to track all(non-batch, batch, chunk) messages, and we can keep the same semantics with the `AckID`.

You can also disscus this issue on the dev mailing list.
BewareMyPower: >  the client doesn't send the ack request to the broker

There is no way to let the client know whether the ack request was sent. That's the point. In addition, users should never care if the ack was sent. They only care if the ack succeeded.

> I remember the Java client has the same behavior

Yeah, but this behavior is wrong. I don't want to blame the author but the error handling is really not taken care of. Happy path is always easy to write.

> because you ignore the ackTracer, so it is confusing.

Oh I see. Let me think a better way to reuse the code.

> You can also disscus this issue on the dev mailing list.

I plan to write a PIP to explain these things, including why to enable batch index ACK by default, the mess of acknowledge APIs' semantics.
BewareMyPower: @nodece I reused the tracking message ID's tracker and used the semantics that ACK on an incomplete message ID in the batch does not return an error. Please check the latest tests. As for reusing the `ackIDCommon` method. It's still hard. There is not much to reuse except for the `ack` method call on the tracking message ID. The `ack` and `ackWithTxn` methods of `unAckChunksTracker` are also a mess.
BewareMyPower: @nodece I reused the tracking message ID's tracker and used the semantics that ACK on an incomplete message ID in the batch does not return an error. Please check the latest tests. As for reusing the `ackIDCommon` method. It's still hard. There is not much to reuse except for the `ack` method call on the tracking message ID. The `ack` and `ackWithTxn` methods of `unAckChunksTracker` are also a mess.
BewareMyPower: @RobertIndie @nodece @crossoverJie @shibd Thanks for your reviews. The `AckIDList` API is updated now.

For the simple consumer case that only 1 topic is subscribed, since only 1 ACK request will be sent, these message IDs either all fail or all succeed. The only exceptional case is that invalid message IDs are passed. However, in this case, there is no way to handle these message IDs because acknowledging them will always fail. Hence I just return a trivial `error` here. 

However, if the consumer subscribes multiple topics, there might be multiple `partitionConsumer` objects here, each of them has a separated events channel (`eventsCh`) so multiple ACK requests could be sent. Therefore, an `AckError` will be returned, which contains a map that maps the failed message ID to its error.

See the API documents and the tests for details. PTAL again.

Patch:
diff --git a/pulsar/ack_grouping_tracker.go b/pulsar/ack_grouping_tracker.go
index c4ecc00..bb9059a 100644
--- a/pulsar/ack_grouping_tracker.go
+++ b/pulsar/ack_grouping_tracker.go
@@ -62,7 +62,7 @@ func newAckGroupingTracker(options *AckGroupingOptions,
 		maxNumAcks:        int(options.MaxSize),
 		ackCumulative:     ackCumulative,
 		ackList:           ackList,
-		pendingAcks:       make(map[[2]uint64]*bitset.BitSet),
+		pendingAcks:       make(map[position]*bitset.BitSet),
 		lastCumulativeAck: EarliestMessageID(),
 	}
 
@@ -110,6 +110,15 @@ func (i *immediateAckGroupingTracker) flushAndClean() {
 func (i *immediateAckGroupingTracker) close() {
 }
 
+type position struct {
+	ledgerID uint64
+	entryID  uint64
+}
+
+func newPosition(msgID MessageID) position {
+	return position{ledgerID: uint64(msgID.LedgerID()), entryID: uint64(msgID.EntryID())}
+}
+
 type timedAckGroupingTracker struct {
 	sync.RWMutex
 
@@ -124,7 +133,7 @@ type timedAckGroupingTracker struct {
 	// 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[[2]uint64]*bitset.BitSet
+	pendingAcks map[position]*bitset.BitSet
 
 	lastCumulativeAck     MessageID
 	cumulativeAckRequired int32
@@ -138,35 +147,36 @@ func (t *timedAckGroupingTracker) add(id MessageID) {
 	}
 }
 
-func (t *timedAckGroupingTracker) tryAddIndividual(id MessageID) map[[2]uint64]*bitset.BitSet {
-	t.Lock()
-	defer t.Unlock()
-	key := [2]uint64{uint64(id.LedgerID()), uint64(id.EntryID())}
-
+func addMsgIDToPendingAcks(pendingAcks map[position]*bitset.BitSet, id MessageID) {
+	key := newPosition(id)
 	batchIdx := id.BatchIdx()
 	batchSize := id.BatchSize()
 
 	if batchIdx >= 0 && batchSize > 0 {
-		bs, found := t.pendingAcks[key]
+		bs, found := pendingAcks[key]
 		if !found {
-			if batchSize > 1 {
-				bs = bitset.New(uint(batchSize))
-				for i := uint(0); i < uint(batchSize); i++ {
-					bs.Set(i)
-				}
+			bs = bitset.New(uint(batchSize))
+			for i := uint(0); i < uint(batchSize); i++ {
+				bs.Set(i)
 			}
-			t.pendingAcks[key] = bs
+			pendingAcks[key] = bs
 		}
 		if bs != nil {
 			bs.Clear(uint(batchIdx))
 		}
 	} else {
-		t.pendingAcks[key] = nil
+		pendingAcks[key] = nil
 	}
+}
 
+func (t *timedAckGroupingTracker) tryAddIndividual(id MessageID) map[position]*bitset.BitSet {
+	t.Lock()
+	defer t.Unlock()
+
+	addMsgIDToPendingAcks(t.pendingAcks, id)
 	if len(t.pendingAcks) >= t.maxNumAcks {
 		pendingAcks := t.pendingAcks
-		t.pendingAcks = make(map[[2]uint64]*bitset.BitSet)
+		t.pendingAcks = make(map[position]*bitset.BitSet)
 		return pendingAcks
 	}
 	return nil
@@ -195,7 +205,7 @@ func (t *timedAckGroupingTracker) isDuplicate(id MessageID) bool {
 	if messageIDCompare(t.lastCumulativeAck, id) >= 0 {
 		return true
 	}
-	key := [2]uint64{uint64(id.LedgerID()), uint64(id.EntryID())}
+	key := newPosition(id)
 	if bs, found := t.pendingAcks[key]; found {
 		if bs == nil {
 			return true
@@ -232,11 +242,11 @@ func (t *timedAckGroupingTracker) flushAndClean() {
 	}
 }
 
-func (t *timedAckGroupingTracker) clearPendingAcks() map[[2]uint64]*bitset.BitSet {
+func (t *timedAckGroupingTracker) clearPendingAcks() map[position]*bitset.BitSet {
 	t.Lock()
 	defer t.Unlock()
 	pendingAcks := t.pendingAcks
-	t.pendingAcks = make(map[[2]uint64]*bitset.BitSet)
+	t.pendingAcks = make(map[position]*bitset.BitSet)
 	return pendingAcks
 }
 
@@ -250,12 +260,10 @@ func (t *timedAckGroupingTracker) close() {
 	}
 }
 
-func (t *timedAckGroupingTracker) flushIndividual(pendingAcks map[[2]uint64]*bitset.BitSet) {
+func toMsgIDDataList(pendingAcks map[position]*bitset.BitSet) []*pb.MessageIdData {
 	msgIDs := make([]*pb.MessageIdData, 0, len(pendingAcks))
 	for k, v := range pendingAcks {
-		ledgerID := k[0]
-		entryID := k[1]
-		msgID := &pb.MessageIdData{LedgerId: &ledgerID, EntryId: &entryID}
+		msgID := &pb.MessageIdData{LedgerId: &k.ledgerID, EntryId: &k.entryID}
 		if v != nil && !v.None() {
 			bytes := v.Bytes()
 			msgID.AckSet = make([]int64, len(bytes))
@@ -265,5 +273,9 @@ func (t *timedAckGroupingTracker) flushIndividual(pendingAcks map[[2]uint64]*bit
 		}
 		msgIDs = append(msgIDs, msgID)
 	}
-	t.ackList(msgIDs)
+	return msgIDs
+}
+
+func (t *timedAckGroupingTracker) flushIndividual(pendingAcks map[position]*bitset.BitSet) {
+	t.ackList(toMsgIDDataList(pendingAcks))
 }
diff --git a/pulsar/consumer.go b/pulsar/consumer.go
index 880cad5..7aee964 100644
--- a/pulsar/consumer.go
+++ b/pulsar/consumer.go
@@ -19,6 +19,8 @@ package pulsar
 
 import (
 	"context"
+	"fmt"
+	"strings"
 	"time"
 
 	"github.com/apache/pulsar-client-go/pulsar/backoff"
@@ -266,6 +268,23 @@ type ConsumerOptions struct {
 	startMessageID *trackingMessageID
 }
 
+// This error is returned when `AckIDList` failed and `AckWithResponse` is true.
+// It only contains the valid message IDs that failed to be acknowledged in the `AckIDList` call.
+// For those invalid message IDs, users should ignore them and not acknowledge them again.
+type AckError map[MessageID]error
+
+func (e AckError) Error() string {
+	builder := strings.Builder{}
+	errorMap := make(map[string][]MessageID)
+	for id, err := range e {
+		errorMap[err.Error()] = append(errorMap[err.Error()], id)
+	}
+	for err, msgIDs := range errorMap {
+		builder.WriteString(fmt.Sprintf("error: %s, failed message IDs: %v\n", err, msgIDs))
+	}
+	return builder.String()
+}
+
 // Consumer is an interface that abstracts behavior of Pulsar's consumer
 type Consumer interface {
 	// Subscription get a subscription for the consumer
@@ -305,8 +324,20 @@ type Consumer interface {
 	Ack(Message) error
 
 	// AckID the consumption of a single message, identified by its MessageID
+	// When `EnableBatchIndexAcknowledgment` is false, if a message ID represents a message in the batch,
+	// it will not be actually acknowledged by broker until all messages in that batch are acknowledged via
+	// `AckID` or `AckIDList`.
 	AckID(MessageID) error
 
+	// AckIDList the consumption of a list of messages, identified by their MessageIDs
+	//
+	// This method should be used when `AckWithResponse` is true. Otherwise, it will be equivalent with calling
+	// `AckID` on each message ID in the list.
+	//
+	// When `AckWithResponse` is true, the returned error could be an `AckError` which contains the failed message ID
+	// and the corresponding error.
+	AckIDList([]MessageID) error
+
 	// AckWithTxn the consumption of a single message with a transaction
 	AckWithTxn(Message, Transaction) error
 
diff --git a/pulsar/consumer_impl.go b/pulsar/consumer_impl.go
index 92376ac..eafa4b4 100644
--- a/pulsar/consumer_impl.go
+++ b/pulsar/consumer_impl.go
@@ -41,6 +41,7 @@ const defaultNackRedeliveryDelay = 1 * time.Minute
 type acker interface {
 	// AckID does not handle errors returned by the Broker side, so no need to wait for doneCh to finish.
 	AckID(id MessageID) error
+	AckIDList(msgIDs []MessageID) error
 	AckIDWithResponse(id MessageID) error
 	AckIDWithTxn(msgID MessageID, txn Transaction) error
 	AckIDCumulative(msgID MessageID) error
@@ -559,6 +560,15 @@ func (c *consumer) AckID(msgID MessageID) error {
 	return c.consumers[msgID.PartitionIdx()].AckID(msgID)
 }
 
+func (c *consumer) AckIDList(msgIDs []MessageID) error {
+	return ackIDListFromMultiTopics(c.log, msgIDs, func(msgID MessageID) (acker, error) {
+		if err := c.checkMsgIDPartition(msgID); err != nil {
+			return nil, err
+		}
+		return c.consumers[msgID.PartitionIdx()], nil
+	})
+}
+
 // AckCumulative the reception of all the messages in the stream up to (and including)
 // the provided message, identified by its MessageID
 func (c *consumer) AckCumulative(msg Message) error {
@@ -712,22 +722,29 @@ func (c *consumer) Seek(msgID MessageID) error {
 		return err
 	}
 
-	if err := c.consumers[msgID.PartitionIdx()].Seek(msgID); err != nil {
-		return err
-	}
-
+	consumer := c.consumers[msgID.PartitionIdx()]
+	consumer.pauseDispatchMessage()
 	// clear messageCh
 	for len(c.messageCh) > 0 {
 		<-c.messageCh
 	}
 
-	return nil
+	return consumer.Seek(msgID)
 }
 
 func (c *consumer) SeekByTime(time time.Time) error {
 	c.Lock()
 	defer c.Unlock()
 	var errs error
+
+	for _, cons := range c.consumers {
+		cons.pauseDispatchMessage()
+	}
+	// clear messageCh
+	for len(c.messageCh) > 0 {
+		<-c.messageCh
+	}
+
 	// run SeekByTime on every partition of topic
 	for _, cons := range c.consumers {
 		if err := cons.SeekByTime(time); err != nil {
@@ -736,11 +753,6 @@ func (c *consumer) SeekByTime(time time.Time) error {
 		}
 	}
 
-	// clear messageCh
-	for len(c.messageCh) > 0 {
-		<-c.messageCh
-	}
-
 	return errs
 }
 
diff --git a/pulsar/consumer_multitopic.go b/pulsar/consumer_multitopic.go
index 3030bda..26430ad 100644
--- a/pulsar/consumer_multitopic.go
+++ b/pulsar/consumer_multitopic.go
@@ -167,6 +167,49 @@ func (c *multiTopicConsumer) AckID(msgID MessageID) error {
 	return mid.consumer.AckID(msgID)
 }
 
+func (c *multiTopicConsumer) AckIDList(msgIDs []MessageID) error {
+	return ackIDListFromMultiTopics(c.log, msgIDs, func(msgID MessageID) (acker, error) {
+		if !checkMessageIDType(msgID) {
+			return nil, fmt.Errorf("invalid message id type %T", msgID)
+		}
+		if mid := toTrackingMessageID(msgID); mid != nil && mid.consumer != nil {
+			return mid.consumer, nil
+		}
+		return nil, errors.New("consumer is nil")
+	})
+}
+
+func ackIDListFromMultiTopics(log log.Logger, msgIDs []MessageID, findConsumer func(MessageID) (acker, error)) error {
+	consumerToMsgIDs := make(map[acker][]MessageID)
+	for _, msgID := range msgIDs {
+		if consumer, err := findConsumer(msgID); err == nil {
+			consumerToMsgIDs[consumer] = append(consumerToMsgIDs[consumer], msgID)
+		} else {
+			log.Warnf("Can not find consumer for %v", msgID)
+		}
+	}
+
+	ackError := AckError{}
+	for consumer, ids := range consumerToMsgIDs {
+		if err := consumer.AckIDList(ids); err != nil {
+			if topicAckError := err.(AckError); topicAckError != nil {
+				for id, err := range topicAckError {
+					ackError[id] = err
+				}
+			} else {
+				// It should not reach here
+				for _, id := range ids {
+					ackError[id] = err
+				}
+			}
+		}
+	}
+	if len(ackError) == 0 {
+		return nil
+	}
+	return ackError
+}
+
 // AckWithTxn the consumption of a single message with a transaction
 func (c *multiTopicConsumer) AckWithTxn(msg Message, txn Transaction) error {
 	msgID := msg.ID()
diff --git a/pulsar/consumer_partition.go b/pulsar/consumer_partition.go
index 0e8274e..471d45a 100644
--- a/pulsar/consumer_partition.go
+++ b/pulsar/consumer_partition.go
@@ -24,6 +24,7 @@ import (
 	"math"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/apache/pulsar-client-go/pulsar/backoff"
@@ -185,6 +186,20 @@ type partitionConsumer struct {
 
 	redirectedClusterURI string
 	backoffPolicyFunc    func() backoff.Policy
+
+	dispatcherSeekingControlCh chan struct{}
+	isSeeking                  atomic.Bool
+}
+
+// pauseDispatchMessage used to discard the message in the dispatcher goroutine.
+// This method will be called When the parent consumer performs the seek operation.
+// After the seek operation, the dispatcher will continue dispatching messages automatically.
+func (pc *partitionConsumer) pauseDispatchMessage() {
+	pc.dispatcherSeekingControlCh <- struct{}{}
+}
+
+func (pc *partitionConsumer) Topic() string {
+	return pc.topic
 }
 
 func (pc *partitionConsumer) ActiveConsumerChanged(isActive bool) {
@@ -329,27 +344,28 @@ func newPartitionConsumer(parent Consumer, client *client, options *partitionCon
 	}
 
 	pc := &partitionConsumer{
-		parentConsumer:       parent,
-		client:               client,
-		options:              options,
-		topic:                options.topic,
-		name:                 options.consumerName,
-		consumerID:           client.rpcClient.NewConsumerID(),
-		partitionIdx:         int32(options.partitionIdx),
-		eventsCh:             make(chan interface{}, 10),
-		maxQueueSize:         int32(options.receiverQueueSize),
-		queueCh:              make(chan *message, options.receiverQueueSize),
-		startMessageID:       atomicMessageID{msgID: options.startMessageID},
-		connectedCh:          make(chan struct{}),
-		messageCh:            messageCh,
-		connectClosedCh:      make(chan *connectionClosed, 1),
-		closeCh:              make(chan struct{}),
-		clearQueueCh:         make(chan func(id *trackingMessageID)),
-		compressionProviders: sync.Map{},
-		dlq:                  dlq,
-		metrics:              metrics,
-		schemaInfoCache:      newSchemaInfoCache(client, options.topic),
-		backoffPolicyFunc:    boFunc,
+		parentConsumer:             parent,
+		client:                     client,
+		options:                    options,
+		topic:                      options.topic,
+		name:                       options.consumerName,
+		consumerID:                 client.rpcClient.NewConsumerID(),
+		partitionIdx:               int32(options.partitionIdx),
+		eventsCh:                   make(chan interface{}, 10),
+		maxQueueSize:               int32(options.receiverQueueSize),
+		queueCh:                    make(chan *message, options.receiverQueueSize),
+		startMessageID:             atomicMessageID{msgID: options.startMessageID},
+		connectedCh:                make(chan struct{}),
+		messageCh:                  messageCh,
+		connectClosedCh:            make(chan *connectionClosed, 1),
+		closeCh:                    make(chan struct{}),
+		clearQueueCh:               make(chan func(id *trackingMessageID)),
+		compressionProviders:       sync.Map{},
+		dlq:                        dlq,
+		metrics:                    metrics,
+		schemaInfoCache:            newSchemaInfoCache(client, options.topic),
+		backoffPolicyFunc:          boFunc,
+		dispatcherSeekingControlCh: make(chan struct{}),
 	}
 	if pc.options.autoReceiverQueueSize {
 		pc.currentQueueSize.Store(initialReceiverQueueSize)
@@ -363,7 +379,12 @@ func newPartitionConsumer(parent Consumer, client *client, options *partitionCon
 	pc.ackGroupingTracker = newAckGroupingTracker(options.ackGroupingOptions,
 		func(id MessageID) { pc.sendIndividualAck(id) },
 		func(id MessageID) { pc.sendCumulativeAck(id) },
-		func(ids []*pb.MessageIdData) { pc.eventsCh <- ids })
+		func(ids []*pb.MessageIdData) {
+			pc.eventsCh <- &ackListRequest{
+				errCh:  nil, // ignore the error
+				msgIDs: ids,
+			}
+		})
 	pc.setConsumerState(consumerInit)
 	pc.log = client.log.SubLogger(log.Fields{
 		"name":         pc.name,
@@ -683,6 +704,86 @@ func (pc *partitionConsumer) AckID(msgID MessageID) error {
 	return pc.ackID(msgID, false)
 }
 
+func (pc *partitionConsumer) AckIDList(msgIDs []MessageID) error {
+	if !pc.options.ackWithResponse {
+		for _, msgID := range msgIDs {
+			if err := pc.AckID(msgID); err != nil {
+				return err
+			}
+		}
+		return nil
+	}
+
+	chunkedMsgIDs := make([]*chunkMessageID, 0) // we need to remove them after acknowledging
+	pendingAcks := make(map[position]*bitset.BitSet)
+	validMsgIDs := make([]MessageID, 0, len(msgIDs))
+
+	// They might be complete after the whole for loop
+	for _, msgID := range msgIDs {
+		if msgID.PartitionIdx() != pc.partitionIdx {
+			pc.log.Errorf("%v inconsistent partition index %v (current: %v)", msgID, msgID.PartitionIdx(), pc.partitionIdx)
+		} else if msgID.BatchIdx() >= 0 && msgID.BatchSize() > 0 &&
+			msgID.BatchIdx() >= msgID.BatchSize() {
+			pc.log.Errorf("%v invalid batch index %v (size: %v)", msgID, msgID.BatchIdx(), msgID.BatchSize())
+		} else {
+			valid := true
+			switch convertedMsgID := msgID.(type) {
+			case *trackingMessageID:
+				position := newPosition(msgID)
+				if convertedMsgID.ack() {
+					pendingAcks[position] = nil
+				} else if pc.options.enableBatchIndexAck {
+					pendingAcks[position] = convertedMsgID.tracker.getAckBitSet()
+				}
+			case *chunkMessageID:
+				for _, id := range pc.unAckChunksTracker.get(convertedMsgID) {
+					pendingAcks[newPosition(id)] = nil
+				}
+				chunkedMsgIDs = append(chunkedMsgIDs, convertedMsgID)
+			case *messageID:
+				pendingAcks[newPosition(msgID)] = nil
+			default:
+				pc.log.Errorf("invalid message id type %T: %v", msgID, msgID)
+				valid = false
+			}
+			if valid {
+				validMsgIDs = append(validMsgIDs, msgID)
+			}
+		}
+	}
+
+	if state := pc.getConsumerState(); state == consumerClosed || state == consumerClosing {
+		pc.log.WithField("state", state).Error("Failed to ack by closing or closed consumer")
+		return toAckError(map[error][]MessageID{errors.New("consumer state is closed"): validMsgIDs})
+	}
+
+	req := &ackListRequest{
+		errCh:  make(chan error),
+		msgIDs: toMsgIDDataList(pendingAcks),
+	}
+	pc.eventsCh <- req
+	if err := <-req.errCh; err != nil {
+		return toAckError(map[error][]MessageID{err: validMsgIDs})
+	}
+	for _, id := range chunkedMsgIDs {
+		pc.unAckChunksTracker.remove(id)
+	}
+	for _, id := range msgIDs {
+		pc.options.interceptors.OnAcknowledge(pc.parentConsumer, id)
+	}
+	return nil
+}
+
+func toAckError(errorMap map[error][]MessageID) AckError {
+	e := AckError{}
+	for err, ids := range errorMap {
+		for _, id := range ids {
+			e[id] = err
+		}
+	}
+	return e
+}
+
 func (pc *partitionConsumer) AckIDCumulative(msgID MessageID) error {
 	if !checkMessageIDType(msgID) {
 		pc.log.Errorf("invalid message id type %T", msgID)
@@ -1015,11 +1116,22 @@ func (pc *partitionConsumer) internalAck(req *ackRequest) {
 	}
 }
 
-func (pc *partitionConsumer) internalAckList(msgIDs []*pb.MessageIdData) {
+func (pc *partitionConsumer) internalAckList(request *ackListRequest) {
+	if request.errCh != nil {
+		reqID := pc.client.rpcClient.NewRequestID()
+		_, err := pc.client.rpcClient.RequestOnCnx(pc._getConn(), reqID, pb.BaseCommand_ACK, &pb.CommandAck{
+			AckType:    pb.CommandAck_Individual.Enum(),
+			ConsumerId: proto.Uint64(pc.consumerID),
+			MessageId:  request.msgIDs,
+			RequestId:  &reqID,
+		})
+		request.errCh <- err
+		return
+	}
 	pc.client.rpcClient.RequestOnCnxNoWait(pc._getConn(), pb.BaseCommand_ACK, &pb.CommandAck{
 		AckType:    pb.CommandAck_Individual.Enum(),
 		ConsumerId: proto.Uint64(pc.consumerID),
-		MessageId:  msgIDs,
+		MessageId:  request.msgIDs,
 	})
 }
 
@@ -1440,17 +1552,20 @@ func (pc *partitionConsumer) dispatcher() {
 			}
 			nextMessageSize = queueMsg.size()
 
-			if pc.dlq.shouldSendToDlq(&nextMessage) {
-				// pass the message to the DLQ router
-				pc.metrics.DlqCounter.Inc()
-				messageCh = pc.dlq.Chan()
+			if !pc.isSeeking.Load() {
+				if pc.dlq.shouldSendToDlq(&nextMessage) {
+					// pass the message to the DLQ router
+					pc.metrics.DlqCounter.Inc()
+					messageCh = pc.dlq.Chan()
+				} else {
+					// pass the message to application channel
+					messageCh = pc.messageCh
+				}
+				pc.metrics.PrefetchedMessages.Dec()
+				pc.metrics.PrefetchedBytes.Sub(float64(len(queueMsg.payLoad)))
 			} else {
-				// pass the message to application channel
-				messageCh = pc.messageCh
+				pc.log.Debug("skip dispatching messages when seeking")
 			}
-
-			pc.metrics.PrefetchedMessages.Dec()
-			pc.metrics.PrefetchedBytes.Sub(float64(len(queueMsg.payLoad)))
 		} else {
 			queueCh = pc.queueCh
 		}
@@ -1483,6 +1598,13 @@ func (pc *partitionConsumer) dispatcher() {
 				pc.log.WithError(err).Error("unable to send initial permits to broker")
 			}
 
+		case _, ok := <-pc.dispatcherSeekingControlCh:
+			if !ok {
+				return
+			}
+			pc.log.Debug("received dispatcherSeekingControlCh, set isSeek to true")
+			pc.isSeeking.Store(true)
+
 		case msg, ok := <-queueCh:
 			if !ok {
 				return
@@ -1541,6 +1663,11 @@ type ackRequest struct {
 	err     error
 }
 
+type ackListRequest struct {
+	errCh  chan error
+	msgIDs []*pb.MessageIdData
+}
+
 type ackWithTxnRequest struct {
 	doneCh      chan struct{}
 	msgID       trackingMessageID
@@ -1587,27 +1714,21 @@ func (pc *partitionConsumer) runEventsLoop() {
 	}()
 	pc.log.Debug("get into runEventsLoop")
 
-	go func() {
-		for {
-			select {
-			case <-pc.closeCh:
-				pc.log.Info("close consumer, exit reconnect")
-				return
-			case connectionClosed := <-pc.connectClosedCh:
-				pc.log.Debug("runEventsLoop will reconnect")
-				pc.reconnectToBroker(connectionClosed)
-			}
-		}
-	}()
-
 	for {
-		for i := range pc.eventsCh {
-			switch v := i.(type) {
+		select {
+		case <-pc.closeCh:
+			pc.log.Info("close consumer, exit reconnect")
+			return
+		case connectionClosed := <-pc.connectClosedCh:
+			pc.log.Debug("runEventsLoop will reconnect")
+			pc.reconnectToBroker(connectionClosed)
+		case event := <-pc.eventsCh:
+			switch v := event.(type) {
 			case *ackRequest:
 				pc.internalAck(v)
 			case *ackWithTxnRequest:
 				pc.internalAckWithTxn(v)
-			case []*pb.MessageIdData:
+			case *ackListRequest:
 				pc.internalAckList(v)
 			case *redeliveryRequest:
 				pc.internalRedeliver(v)
@@ -1680,6 +1801,9 @@ func (pc *partitionConsumer) internalClose(req *closeRequest) {
 }
 
 func (pc *partitionConsumer) reconnectToBroker(connectionClosed *connectionClosed) {
+	if pc.isSeeking.CompareAndSwap(true, false) {
+		pc.log.Debug("seek operation triggers reconnection, and reset isSeeking")
+	}
 	var (
 		maxRetry                                    int
 		delayReconnectTime, totalDelayReconnectTime time.Duration
diff --git a/pulsar/consumer_regex.go b/pulsar/consumer_regex.go
index 58cfa80..ced770e 100644
--- a/pulsar/consumer_regex.go
+++ b/pulsar/consumer_regex.go
@@ -215,6 +215,18 @@ func (c *regexConsumer) AckID(msgID MessageID) error {
 	return mid.consumer.AckID(msgID)
 }
 
+func (c *regexConsumer) AckIDList(msgIDs []MessageID) error {
+	return ackIDListFromMultiTopics(c.log, msgIDs, func(msgID MessageID) (acker, error) {
+		if !checkMessageIDType(msgID) {
+			return nil, fmt.Errorf("invalid message id type %T", msgID)
+		}
+		if mid := toTrackingMessageID(msgID); mid.consumer != nil {
+			return mid.consumer, nil
+		}
+		return nil, errors.New("consumer is nil in consumer_regex")
+	})
+}
+
 // AckID the consumption of a single message, identified by its MessageID
 func (c *regexConsumer) AckWithTxn(msg Message, txn Transaction) error {
 	msgID := msg.ID()
diff --git a/pulsar/consumer_zero_queue.go b/pulsar/consumer_zero_queue.go
index 8117127..3f2862d 100644
--- a/pulsar/consumer_zero_queue.go
+++ b/pulsar/consumer_zero_queue.go
@@ -171,6 +171,10 @@ func (z *zeroQueueConsumer) AckID(msgID MessageID) error {
 	return z.pc.AckID(msgID)
 }
 
+func (z *zeroQueueConsumer) AckIDList(msgIDs []MessageID) error {
+	return z.pc.AckIDList(msgIDs)
+}
+
 func (z *zeroQueueConsumer) AckWithTxn(msg Message, txn Transaction) error {
 	msgID := msg.ID()
 	if err := z.checkMsgIDPartition(msgID); err != nil {
diff --git a/pulsar/impl_message.go b/pulsar/impl_message.go
index 478b1af..0acd782 100644
--- a/pulsar/impl_message.go
+++ b/pulsar/impl_message.go
@@ -404,6 +404,12 @@ type ackTracker struct {
 	prevBatchAcked uint32
 }
 
+func (t *ackTracker) getAckBitSet() *bitset.BitSet {
+	t.Lock()
+	defer t.Unlock()
+	return t.batchIDs.Clone()
+}
+
 func (t *ackTracker) ack(batchID int) bool {
 	if batchID < 0 {
 		return true


Test patch:
diff --git a/pulsar/consumer_multitopic_test.go b/pulsar/consumer_multitopic_test.go
index 7c6b898..4b3ec90 100644
--- a/pulsar/consumer_multitopic_test.go
+++ b/pulsar/consumer_multitopic_test.go
@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/apache/pulsar-client-go/pulsaradmin"
 	"github.com/apache/pulsar-client-go/pulsaradmin/pkg/admin/config"
@@ -218,3 +219,101 @@ func TestMultiTopicGetLastMessageIDs(t *testing.T) {
 	}
 
 }
+
+func TestMultiTopicAckIDList(t *testing.T) {
+	for _, params := range []bool{true, false} {
+		t.Run(fmt.Sprintf("TestMultiTopicConsumerAckIDList%v", params), func(t *testing.T) {
+			runMultiTopicAckIDList(t, params)
+		})
+	}
+}
+
+func runMultiTopicAckIDList(t *testing.T, regex bool) {
+	topicPrefix := fmt.Sprintf("multiTopicAckIDList%v", time.Now().UnixNano())
+	topic1 := "persistent://public/default/" + topicPrefix + "1"
+	topic2 := "persistent://public/default/" + topicPrefix + "2"
+
+	client, err := NewClient(ClientOptions{URL: "pulsar://localhost:6650"})
+	assert.Nil(t, err)
+	defer client.Close()
+
+	if regex {
+		admin, err := pulsaradmin.NewClient(&config.Config{})
+		assert.Nil(t, err)
+		for _, topic := range []string{topic1, topic2} {
+			topicName, err := utils.GetTopicName(topic)
+			assert.Nil(t, err)
+			admin.Topics().Create(*topicName, 0)
+		}
+	}
+
+	createConsumer := func() Consumer {
+		options := ConsumerOptions{
+			SubscriptionName: "sub",
+			Type:             Shared,
+			AckWithResponse:  true,
+		}
+		if regex {
+			options.TopicsPattern = topicPrefix + ".*"
+		} else {
+			options.Topics = []string{topic1, topic2}
+		}
+		consumer, err := client.Subscribe(options)
+		assert.Nil(t, err)
+		return consumer
+	}
+	consumer := createConsumer()
+
+	sendMessages(t, client, topic1, 0, 3, false)
+	sendMessages(t, client, topic2, 0, 2, false)
+
+	receiveMessageMap := func(consumer Consumer, numMessages int) map[string][]Message {
+		msgs := receiveMessages(t, consumer, numMessages)
+		topicToMsgs := make(map[string][]Message)
+		for _, msg := range msgs {
+			topicToMsgs[msg.Topic()] = append(topicToMsgs[msg.Topic()], msg)
+		}
+		return topicToMsgs
+	}
+
+	topicToMsgs := receiveMessageMap(consumer, 5)
+	assert.Equal(t, 3, len(topicToMsgs[topic1]))
+	for i := 0; i < 3; i++ {
+		assert.Equal(t, fmt.Sprintf("msg-%d", i), string(topicToMsgs[topic1][i].Payload()))
+	}
+	assert.Equal(t, 2, len(topicToMsgs[topic2]))
+	for i := 0; i < 2; i++ {
+		assert.Equal(t, fmt.Sprintf("msg-%d", i), string(topicToMsgs[topic2][i].Payload()))
+	}
+
+	assert.Nil(t, consumer.AckIDList([]MessageID{
+		topicToMsgs[topic1][0].ID(),
+		topicToMsgs[topic1][2].ID(),
+		topicToMsgs[topic2][1].ID(),
+	}))
+
+	consumer.Close()
+	consumer = createConsumer()
+	topicToMsgs = receiveMessageMap(consumer, 2)
+	assert.Equal(t, 1, len(topicToMsgs[topic1]))
+	assert.Equal(t, "msg-1", string(topicToMsgs[topic1][0].Payload()))
+	assert.Equal(t, 1, len(topicToMsgs[topic2]))
+	assert.Equal(t, "msg-0", string(topicToMsgs[topic2][0].Payload()))
+	consumer.Close()
+
+	msgID0 := topicToMsgs[topic1][0].ID()
+	err = consumer.AckIDList([]MessageID{msgID0})
+	assert.NotNil(t, err)
+	t.Logf("AckIDList error: %v", err)
+
+	msgID1 := topicToMsgs[topic2][0].ID()
+	if ackError, ok := consumer.AckIDList([]MessageID{msgID0, msgID1}).(AckError); ok {
+		assert.Equal(t, 2, len(ackError))
+		assert.Contains(t, ackError, msgID0)
+		assert.Equal(t, "consumer state is closed", ackError[msgID0].Error())
+		assert.Contains(t, ackError, msgID1)
+		assert.Equal(t, "consumer state is closed", ackError[msgID1].Error())
+	} else {
+		assert.Fail(t, "AckIDList should return AckError")
+	}
+}
diff --git a/pulsar/consumer_test.go b/pulsar/consumer_test.go
index d8f3145..2524f68 100644
--- a/pulsar/consumer_test.go
+++ b/pulsar/consumer_test.go
@@ -4745,3 +4745,135 @@ func TestLookupConsumer(t *testing.T) {
 		consumer.Ack(msg)
 	}
 }
+
+func TestAckIDList(t *testing.T) {
+	for _, params := range []bool{true, false} {
+		t.Run(fmt.Sprintf("TestAckIDList_%v", params), func(t *testing.T) {
+			runAckIDListTest(t, params)
+		})
+	}
+}
+
+func runAckIDListTest(t *testing.T, enableBatchIndexAck bool) {
+	client, err := NewClient(ClientOptions{URL: lookupURL})
+	assert.Nil(t, err)
+	defer client.Close()
+
+	topic := fmt.Sprintf("test-ack-id-list-%v", time.Now().Nanosecond())
+
+	consumer := createSharedConsumer(t, client, topic, enableBatchIndexAck)
+	sendMessages(t, client, topic, 0, 5, true)  // entry 0: [0, 1, 2, 3, 4]
+	sendMessages(t, client, topic, 5, 3, false) // entry 2: [5], 3: [6], 4: [7]
+	sendMessages(t, client, topic, 8, 2, true)  // entry 5: [8, 9]
+
+	msgs := receiveMessages(t, consumer, 10)
+	originalMsgIDs := make([]MessageID, 0)
+	for i := 0; i < 10; i++ {
+		originalMsgIDs = append(originalMsgIDs, msgs[i].ID())
+		assert.Equal(t, fmt.Sprintf("msg-%d", i), string(msgs[i].Payload()))
+	}
+
+	ackedIndexes := []int{0, 2, 3, 6, 8, 9}
+	unackedIndexes := []int{1, 4, 5, 7}
+	if !enableBatchIndexAck {
+		// [0, 4] is the first batch range but only partial of it is acked
+		unackedIndexes = []int{0, 1, 2, 3, 4, 5, 7}
+	}
+	msgIDs := make([]MessageID, len(ackedIndexes))
+	for i := 0; i < len(ackedIndexes); i++ {
+		msgIDs[i] = msgs[ackedIndexes[i]].ID()
+	}
+	assert.Nil(t, consumer.AckIDList(msgIDs))
+	consumer.Close()
+
+	consumer = createSharedConsumer(t, client, topic, enableBatchIndexAck)
+	msgs = receiveMessages(t, consumer, len(unackedIndexes))
+	for i := 0; i < len(unackedIndexes); i++ {
+		assert.Equal(t, fmt.Sprintf("msg-%d", unackedIndexes[i]), string(msgs[i].Payload()))
+	}
+
+	if !enableBatchIndexAck {
+		msgIDs = make([]MessageID, 0)
+		for i := 0; i < 5; i++ {
+			msgIDs = append(msgIDs, originalMsgIDs[i])
+		}
+		assert.Nil(t, consumer.AckIDList(msgIDs))
+		consumer.Close()
+
+		consumer = createSharedConsumer(t, client, topic, enableBatchIndexAck)
+		msgs = receiveMessages(t, consumer, 2)
+		assert.Equal(t, "msg-5", string(msgs[0].Payload()))
+		assert.Equal(t, "msg-7", string(msgs[1].Payload()))
+		consumer.Close()
+	}
+	consumer.Close()
+	err = consumer.AckIDList(msgIDs)
+	assert.NotNil(t, err)
+	if ackError := err.(AckError); ackError != nil {
+		assert.Equal(t, len(msgIDs), len(ackError))
+		for _, id := range msgIDs {
+			assert.Contains(t, ackError, id)
+			assert.Equal(t, "consumer state is closed", ackError[id].Error())
+		}
+	} else {
+		assert.Fail(t, "AckIDList should return AckError")
+	}
+}
+
+func createSharedConsumer(t *testing.T, client Client, topic string, enableBatchIndexAck bool) Consumer {
+	consumer, err := client.Subscribe(ConsumerOptions{
+		Topic:                          topic,
+		SubscriptionName:               "my-sub",
+		SubscriptionInitialPosition:    SubscriptionPositionEarliest,
+		Type:                           Shared,
+		EnableBatchIndexAcknowledgment: enableBatchIndexAck,
+		AckWithResponse:                true,
+	})
+	assert.Nil(t, err)
+	return consumer
+}
+
+func sendMessages(t *testing.T, client Client, topic string, startIndex int, numMessages int, batching bool) {
+	producer, err := client.CreateProducer(ProducerOptions{
+		Topic:                   topic,
+		DisableBatching:         !batching,
+		BatchingMaxMessages:     uint(numMessages),
+		BatchingMaxSize:         1024 * 1024 * 10,
+		BatchingMaxPublishDelay: 1 * time.Hour,
+	})
+	assert.Nil(t, err)
+	defer producer.Close()
+
+	ctx := context.Background()
+	for i := 0; i < numMessages; i++ {
+		msg := &ProducerMessage{Payload: []byte(fmt.Sprintf("msg-%d", startIndex+i))}
+		if batching {
+			producer.SendAsync(ctx, msg, func(_ MessageID, _ *ProducerMessage, err error) {
+				if err != nil {
+					t.Logf("Failed to send message: %v", err)
+				}
+			})
+		} else {
+			if _, err := producer.Send(ctx, msg); err != nil {
+				assert.Fail(t, "Failed to send message: %v", err)
+			}
+		}
+	}
+	assert.Nil(t, producer.Flush())
+}
+
+func receiveMessages(t *testing.T, consumer Consumer, numMessages int) []Message {
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	msgs := make([]Message, 0)
+	for i := 0; i < numMessages; i++ {
+		if msg, err := consumer.Receive(ctx); err == nil {
+			msgs = append(msgs, msg)
+		} else {
+			t.Logf("Failed to receive message: %v", err)
+			break
+		}
+	}
+	assert.Equal(t, numMessages, len(msgs))
+	return msgs
+}
diff --git a/pulsar/internal/pulsartracing/consumer_interceptor_test.go b/pulsar/internal/pulsartracing/consumer_interceptor_test.go
index 1fa1bf0..e771235 100644
--- a/pulsar/internal/pulsartracing/consumer_interceptor_test.go
+++ b/pulsar/internal/pulsartracing/consumer_interceptor_test.go
@@ -79,6 +79,10 @@ func (c *mockConsumer) AckID(_ pulsar.MessageID) error {
 	return nil
 }
 
+func (c *mockConsumer) AckIDList(_ []pulsar.MessageID) error {
+	return nil
+}
+
 func (c *mockConsumer) AckCumulative(_ pulsar.Message) error {
 	return nil
 }