[fix][broker] fix incorrect unack stats when dup ack a message

[labuladong]

Fixes #19268

Modifications

For shared-type consumers, if can't find the message position in pendingAck set, skip the following process.

Verifying this change

• Make sure that the change passes the CI checks.

Does this pull request potentially affect one of the following parts:

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

Matching PR in forked repository

PR in forked repository: labuladong#20

[codelipenghui]

The ackOwnerConsumer can be a different consumer.

private Consumer getAckOwnerConsumer(long ledgerId, long entryId) {
    Consumer ackOwnerConsumer = this;
    if (Subscription.isIndividualAckMode(subType)) {
        if (!getPendingAcks().containsKey(ledgerId, entryId)) {
            for (Consumer consumer : subscription.getConsumers()) {
                if (consumer != this && consumer.getPendingAcks().containsKey(ledgerId, entryId)) {
                    ackOwnerConsumer = consumer;
                    break;
                }
            }
        }
    }
    return ackOwnerConsumer;
}

We will still have the issue that the unacked messages count is decreased for this consumer, but actually the owner consumer is another consumer.

[labuladong]

I don't understand, if the actual consumer is another consumer, addAndGetUnAckedMsgs can update the unack count of that consumer:

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

Lines 1049 to 1060 in 96fb7da

	private int addAndGetUnAckedMsgs(Consumer consumer, int ackedMessages) {
	int unackedMsgs = 0;
	if (Subscription.isIndividualAckMode(subType)) {
	subscription.addUnAckedMessages(ackedMessages);
	unackedMsgs = UNACKED_MESSAGES_UPDATER.addAndGet(consumer, ackedMessages);
	}
	if (unackedMsgs < 0 && System.currentTimeMillis() - negtiveUnackedMsgsTimestamp >= 10_000) {
	negtiveUnackedMsgsTimestamp = System.currentTimeMillis();
	log.warn("unackedMsgs is : {}, ackedMessages : {}, consumer : {}", unackedMsgs, ackedMessages, consumer);
	}
	return unackedMsgs;
	}

[github-actions]

The pr had no activity for 30 days, mark with Stale label.

[poorbarcode]

Since we will start the RC version of 3.0.0 on 2023-04-11, I will change the label/milestone of PR who have not been merged.

• The PR of type feature is deferred to 3.1.0
• The PR of type fix is deferred to 3.0.1

So drag this PR to 3.0.1

[github-actions]

The pr had no activity for 30 days, mark with Stale label.

[nodece]

#19268 has been fixed by #20990, so close this PR.