Fix batch ack count is negtive issue.

[Technoboy-]

Motivation

As #13383 fixed the batch ack issue. we find that the unack-msg count could be negative(#14246). At first, we think it was the normal case caused by msg redelivery. But after diving into the logic, we find it's a bug.

The test is copy from #14246 :

for (int i = 0; i < 50; i++) {
      Message<String> msg = consumer.receive();
      if (i % 2 == 0) {
           consumer.acknowledgeAsync(msg);
       } else {
            consumer.negativeAcknowledge(msg);
       }
}       

When msg is negativeAcknowledge, Consumer#redeliverUnacknowledgedMessages will invoke:

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java

Lines 900 to 912 in b22f706

	public void redeliverUnacknowledgedMessages(List<MessageIdData> messageIds) {
	int totalRedeliveryMessages = 0;
	List<PositionImpl> pendingPositions = Lists.newArrayList();
	for (MessageIdData msg : messageIds) {
	PositionImpl position = PositionImpl.get(msg.getLedgerId(), msg.getEntryId());
	LongPair longPair = pendingAcks.get(position.getLedgerId(), position.getEntryId());
	if (longPair != null) {
	int unAckedCount = (int) getUnAckedCountForBatchIndexLevelEnabled(position, longPair.first);
	pendingAcks.remove(position.getLedgerId(), position.getEntryId());
	totalRedeliveryMessages += unAckedCount;
	pendingPositions.add(position);
	}
	}

When calculating totalRedeliveryMessages, it must check pendingAcks contains this message. and remove from pendingAcks after that. (Dispatch messages will add messages to pendingAcks)
So the above test may exist that when negativeAcknowledge first and then acknowledgeAsync.
acknowledgeAsync mapped to Consumer#individualAckNormal and decrease unack-msg in :

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java

Lines 543 to 561 in b22f706

	private long getAckedCountForBatchIndexLevelEnabled(PositionImpl position, long batchSize, long[] ackSets) {
	long ackedCount = 0;
	if (isAcknowledgmentAtBatchIndexLevelEnabled && Subscription.isIndividualAckMode(subType)) {
	long[] cursorAckSet = getCursorAckSet(position);
	if (cursorAckSet != null) {
	BitSetRecyclable cursorBitSet = BitSetRecyclable.create().resetWords(cursorAckSet);
	int lastCardinality = cursorBitSet.cardinality();
	BitSetRecyclable givenBitSet = BitSetRecyclable.create().resetWords(ackSets);
	cursorBitSet.and(givenBitSet);
	givenBitSet.recycle();
	int currentCardinality = cursorBitSet.cardinality();
	ackedCount = lastCardinality - currentCardinality;
	cursorBitSet.recycle();
	} else if (pendingAcks.get(position.getLedgerId(), position.getEntryId()) != null) {
	ackedCount = batchSize - BitSet.valueOf(ackSets).cardinality();
	}
	}
	return ackedCount;
	}

It doesn't check pendingAcks. this is the root cause. Should move line 556 to 545.

Documentation

• no-need-doc