Don't increment unacked messages for the consumer with Exclusive/Fail…

…over subscription mode. (apache#6558)

Fixes apache#6552

### Motivation

apache#6552 is introduced by apache#5929, so this PR stop increase unacked messages for the consumer with Exclusive/Failover subscription mode.

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

@@ -282,7 +282,9 @@ public ChannelPromise sendMessages(final List<Entry> entries, EntryBatchSizes ba
     }
 
     private void incrementUnackedMessages(int ackedMessages) {
-        if (addAndGetUnAckedMsgs(this, ackedMessages) >= maxUnackedMessages && shouldBlockConsumerOnUnackMsgs()) {
+        if (Subscription.isIndividualAckMode(subType)
+                && addAndGetUnAckedMsgs(this, ackedMessages) >= maxUnackedMessages
+                && maxUnackedMessages > 0) {
             blockedConsumerOnUnackedMsgs = true;
         }
     }

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

@@ -103,4 +103,41 @@ public void testConsumerStatsOnZeroMaxUnackedMessagesPerConsumer() throws Pulsar
         Assert.assertEquals(stats.subscriptions.entrySet().iterator().next().getValue().consumers.get(0).unackedMessages, 0);
     }
 
+    @Test
+    public void testAckStatsOnPartitionedTopicForExclusiveSubscription() throws PulsarAdminException, PulsarClientException, InterruptedException {
+        final String topic = "persistent://my-property/my-ns/testAckStatsOnPartitionedTopicForExclusiveSubscription";
+        admin.topics().createPartitionedTopic(topic, 3);
+        Consumer<byte[]> consumer = pulsarClient.newConsumer()
+                .topic(topic)
+                .subscriptionType(SubscriptionType.Exclusive)
+                .subscriptionName("sub")
+                .subscribe();
+
+        Producer<byte[]> producer = pulsarClient.newProducer()
+                .topic(topic)
+                .create();
+
+        final int messages = 10;
+        for (int i = 0; i < messages; i++) {
+            producer.send(("message-" + i).getBytes());
+        }
+
+        int received = 0;
+        for (int i = 0; i < messages; i++) {
+            consumer.acknowledge(consumer.receive());
+            received++;
+        }
+        Assert.assertEquals(messages, received);
+
+        // wait acknowledge send
+        Thread.sleep(2000);
+
+        for (int i = 0; i < 3; i++) {
+            TopicStats stats = admin.topics().getStats(topic + "-partition-" + i);
+            Assert.assertEquals(stats.subscriptions.size(), 1);
+            Assert.assertEquals(stats.subscriptions.entrySet().iterator().next().getValue().consumers.size(), 1);
+            Assert.assertEquals(stats.subscriptions.entrySet().iterator().next().getValue().consumers.get(0).unackedMessages, 0);
+        }
+    }
+
 }