Address comments

Author: BewareMyPower

pulsar/consumer_partition.go

@@ -702,7 +702,6 @@ func (pc *partitionConsumer) AckIDList(msgIDs []MessageID) error {
 	pendingAcks := make(map[position]*bitset.BitSet)
 
 	// They might be complete after the whole for loop
-	incompleteTrackingIDs := make([]*trackingMessageID, 0)
 	for _, msgID := range msgIDs {
 		if msgID.PartitionIdx() != pc.partitionIdx {
 			pc.log.Errorf("%v inconsistent partition index %v (current: %v)", msgID, msgID.PartitionIdx(), pc.partitionIdx)
@@ -712,10 +711,11 @@ func (pc *partitionConsumer) AckIDList(msgIDs []MessageID) error {
 		} else {
 			switch convertedMsgID := msgID.(type) {
 			case *trackingMessageID:
+				position := newPosition(msgID)
 				if convertedMsgID.ack() {
-					pendingAcks[newPosition(msgID)] = nil
-				} else {
-					incompleteTrackingIDs = append(incompleteTrackingIDs, convertedMsgID)
+					pendingAcks[position] = nil
+				} else if pc.options.enableBatchIndexAck {
+					pendingAcks[position] = convertedMsgID.tracker.getAckBitSet()
 				}
 			case *chunkMessageID:
 				for _, id := range pc.unAckChunksTracker.get(convertedMsgID) {
@@ -730,15 +730,6 @@ func (pc *partitionConsumer) AckIDList(msgIDs []MessageID) error {
 		}
 	}
 
-	if pc.options.enableBatchIndexAck {
-		for _, trackingID := range incompleteTrackingIDs {
-			position := newPosition(trackingID)
-			if _, found := pendingAcks[position]; !found {
-				pendingAcks[position] = trackingID.tracker.getAckBitSet()
-			}
-		}
-	}
-
 	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")

pulsar/consumer_test.go

@@ -4827,8 +4827,11 @@ func createSharedConsumer(t *testing.T, client Client, topic string, enableBatch
 
 func sendMessages(t *testing.T, client Client, topic string, startIndex int, numMessages int, batching bool) {
 	producer, err := client.CreateProducer(ProducerOptions{
-		Topic:           topic,
-		DisableBatching: !batching,
+		Topic:                   topic,
+		DisableBatching:         !batching,
+		BatchingMaxMessages:     uint(numMessages),
+		BatchingMaxSize:         1024 * 1024 * 10,
+		BatchingMaxPublishDelay: 1 * time.Hour,
 	})
 	assert.Nil(t, err)
 	defer producer.Close()
@@ -4837,12 +4840,18 @@ func sendMessages(t *testing.T, client Client, topic string, startIndex int, num
 	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, _ error) {})
+			producer.SendAsync(ctx, msg, func(_ MessageID, _ *ProducerMessage, err error) {
+				if err != nil {
+					t.Logf("Failed to send message: %v", err)
+				}
+			})
 		} else {
-			producer.Send(ctx, msg)
+			if _, err := producer.Send(ctx, msg); err != nil {
+				assert.Fail(t, "Failed to send message: %v", err)
+			}
 		}
 	}
-	producer.Flush()
+	assert.Nil(t, producer.Flush())
 }
 
 func receiveMessages(t *testing.T, consumer Consumer, numMessages int) []Message {
@@ -4851,10 +4860,9 @@ func receiveMessages(t *testing.T, consumer Consumer, numMessages int) []Message
 	msgs := make([]Message, 0)
 	for i := 0; i < numMessages; i++ {
 		if msg, err := consumer.Receive(ctx); err == nil {
-			fmt.Println("Received message: ", string(msg.Payload()))
 			msgs = append(msgs, msg)
 		} else {
-			fmt.Printf("Failed to receive message: %v", err)
+			t.Logf("Failed to receive message: %v", err)
 			break
 		}
 	}