[fix][broker] Fix incorrect unack count when using shared subscriptio…

…n on non-persistent topic (#21592)

Fixes #21568

Motivation
Fix incorrect unack count when using shared subscription on non-persistent topic

Modifications
In the case of a non-persistent topic, the consumer does not send an ack to the broker (see org.apache.pulsar.client.impl.NonPersistentAcknowledgmentGroupingTracker# addAcknowledgment)

To work around this, we can update unackedMessages when the broker sends a message to the consumer successfully.

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

@@ -115,6 +115,9 @@ public class Consumer {
     private final ConsumerStatsImpl stats;
 
     private final boolean isDurable;
+
+    private final boolean isPersistentTopic;
+
     private static final AtomicIntegerFieldUpdater<Consumer> UNACKED_MESSAGES_UPDATER =
             AtomicIntegerFieldUpdater.newUpdater(Consumer.class, "unackedMessages");
     private volatile int unackedMessages = 0;
@@ -172,6 +175,7 @@ public Consumer(Subscription subscription, SubType subType, String topicName, lo
         this.readCompacted = readCompacted;
         this.consumerName = consumerName;
         this.isDurable = isDurable;
+        this.isPersistentTopic = subscription.getTopic() instanceof PersistentTopic;
         this.keySharedMeta = keySharedMeta;
         this.cnx = cnx;
         this.msgOut = new Rate();
@@ -239,6 +243,7 @@ public Consumer(Subscription subscription, SubType subType, String topicName, lo
         this.pendingAcks = null;
         this.stats = null;
         this.isDurable = false;
+        this.isPersistentTopic = false;
         this.metadata = null;
         this.keySharedMeta = null;
         this.clientAddress = null;
@@ -1073,7 +1078,7 @@ public Subscription getSubscription() {
 
     private int addAndGetUnAckedMsgs(Consumer consumer, int ackedMessages) {
         int unackedMsgs = 0;
-        if (Subscription.isIndividualAckMode(subType)) {
+        if (isPersistentTopic && Subscription.isIndividualAckMode(subType)) {
             subscription.addUnAckedMessages(ackedMessages);
             unackedMsgs = UNACKED_MESSAGES_UPDATER.addAndGet(consumer, ackedMessages);
         }

pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/ConsumerStatsTest.java

@@ -446,4 +446,37 @@ public void testAvgMessagesPerEntry() throws Exception {
         int avgMessagesPerEntry = consumerStats.getAvgMessagesPerEntry();
         assertEquals(3, avgMessagesPerEntry);
     }
+
+    @Test()
+    public void testNonPersistentTopicSharedSubscriptionUnackedMessages() throws Exception {
+        final String topicName = "non-persistent://my-property/my-ns/my-topic" + UUID.randomUUID();
+        final String subName = "my-sub";
+
+        @Cleanup
+        Producer<byte[]> producer = pulsarClient.newProducer()
+                .topic(topicName)
+                .create();
+        @Cleanup
+        Consumer<byte[]> consumer = pulsarClient.newConsumer()
+                .topic(topicName)
+                .subscriptionName(subName)
+                .subscriptionType(SubscriptionType.Shared)
+                .subscribe();
+
+        for (int i = 0; i < 5; i++) {
+            producer.send(("message-" + i).getBytes());
+        }
+        for (int i = 0; i < 5; i++) {
+            Message<byte[]> msg = consumer.receive(5, TimeUnit.SECONDS);
+            consumer.acknowledge(msg);
+        }
+        TimeUnit.SECONDS.sleep(1);
+
+        TopicStats topicStats = admin.topics().getStats(topicName);
+        assertEquals(1, topicStats.getSubscriptions().size());
+        List<? extends ConsumerStats> consumers = topicStats.getSubscriptions().get(subName).getConsumers();
+        assertEquals(1, consumers.size());
+        assertEquals(0, consumers.get(0).getUnackedMessages());
+    }
+
 }