[fix][broker] Debit un-acked messages only when the consumer is actually removed

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

@@ -4125,6 +4125,11 @@ public ExecutorService getTopicPoliciesNotifyThread(TopicName topicName) {
         return topicOrderedExecutor.chooseThread(baseTopicName);
     }
 
+    @VisibleForTesting
+    long getTotalUnackedMessages() {
+        return totalUnackedMessages.sum();
+    }
+
     /**
      * If per-broker unacked message reached to limit then it blocks dispatcher if its unacked message limit has been
      * reached to {@link #maxUnackedMsgsPerDispatcher}.

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java

@@ -241,9 +241,11 @@ protected boolean isConsumersExceededOnSubscription() {
 
     @Override
     public synchronized void removeConsumer(Consumer consumer) throws BrokerServiceException {
-        // decrement unack-message count for removed consumer
-        addUnAckedMessages(-consumer.getUnackedMessages());
         if (consumerSet.removeAll(consumer) == 1) {
+            // decrement unack-message count for removed consumer. Only the removal that actually
+            // unregisters the consumer may debit it, otherwise removing an already-removed consumer
+            // debits the same messages again and drives the subscription counter negative.
+            addUnAckedMessages(-consumer.getUnackedMessages());
             consumerList.remove(consumer);
             log.info()
                     .attr("consumer", consumer)
@@ -279,6 +281,9 @@ public synchronized void removeConsumer(Consumer consumer) throws BrokerServiceE
              * are not mismatch with {@link #consumerSet}. See more detail: https://github.com/apache/pulsar/pull/22270.
              */
             log.error().attr("consumer", consumer).log("Trying to remove a non-connected consumer");
+            // The debit belongs to the removal that unregisters the consumer; do not repeat it here.
+            // The add-consumer failure path can also unregister via internalRemoveConsumer, but that
+            // consumer has not received any messages and therefore has nothing to debit.
             consumerList.removeIf(c -> consumer.equals(c));
             if (consumerList.isEmpty()) {
                 clearComponentsAfterRemovedAllConsumers();

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumersClassic.java

@@ -231,9 +231,11 @@ protected boolean isConsumersExceededOnSubscription() {
 
     @Override
     public synchronized void removeConsumer(Consumer consumer) throws BrokerServiceException {
-        // decrement unack-message count for removed consumer
-        addUnAckedMessages(-consumer.getUnackedMessages());
         if (consumerSet.removeAll(consumer) == 1) {
+            // decrement unack-message count for removed consumer. Only the removal that actually
+            // unregisters the consumer may debit it, otherwise removing an already-removed consumer
+            // debits the same messages again and drives the subscription counter negative.
+            addUnAckedMessages(-consumer.getUnackedMessages());
             consumerList.remove(consumer);
             log.info()
                     .attr("consumer", consumer)
@@ -264,6 +266,7 @@ public synchronized void removeConsumer(Consumer consumer) throws BrokerServiceE
              * are not mismatch with {@link #consumerSet}. See more detail: https://github.com/apache/pulsar/pull/22270.
              */
             log.error().attr("consumer", consumer).log("Trying to remove a non-connected consumer");
+            // The debit belongs to the removal that unregisters the consumer; do not repeat it here.
             consumerList.removeIf(c -> consumer.equals(c));
             if (consumerList.isEmpty()) {
                 clearComponentsAfterRemovedAllConsumers();

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SharedSubscriptionUnackedMessagesAccountingTest.java

@@ -0,0 +1,129 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import java.util.concurrent.TimeUnit;
+import org.apache.pulsar.broker.service.persistent.AbstractPersistentDispatcherMultipleConsumers;
+import org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers;
+import org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumersClassic;
+import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.ProducerConsumerBase;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Factory;
+import org.testng.annotations.Test;
+
+/**
+ * Consumer removal must debit both subscription and broker unacknowledged-message counters exactly once.
+ * Each dispatcher variant owns its broker so configuration changes cannot affect the shared test cluster.
+ */
+@Test(groups = "broker-api")
+public class SharedSubscriptionUnackedMessagesAccountingTest extends ProducerConsumerBase {
+    private static final String SUBSCRIPTION = "shared-churn-sub";
+    private static final int UNACKED_MESSAGES = 10;
+    private final boolean classic;
+
+    @Factory
+    public static Object[] createTestInstances() {
+        return new Object[] {new SharedSubscriptionUnackedMessagesAccountingTest(false),
+                new SharedSubscriptionUnackedMessagesAccountingTest(true)};
+    }
+
+    public SharedSubscriptionUnackedMessagesAccountingTest(boolean classic) {
+        this.classic = classic;
+    }
+
+    @Override
+    protected void doInitConf() throws Exception {
+        super.doInitConf();
+        conf.setSubscriptionSharedUseClassicPersistentImplementation(classic);
+        conf.setMaxUnackedMessagesPerBroker(1000);
+    }
+
+    @Override
+    @BeforeMethod
+    protected void setup() throws Exception {
+        super.internalSetup();
+        super.producerBaseSetup();
+    }
+
+    @Override
+    @AfterMethod(alwaysRun = true)
+    protected void cleanup() throws Exception {
+        super.internalCleanup();
+    }
+
+    @Test(timeOut = 60_000)
+    public void testRemovingSameConsumerTwiceDebitsUnackedMessagesOnce() throws Exception {
+        String topicName = newTopicName();
+        admin.topics().createNonPartitionedTopic(topicName);
+        try (Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+                     .topic(topicName).enableBatching(false).create();
+             org.apache.pulsar.client.api.Consumer<String> departing = pulsarClient.newConsumer(Schema.STRING)
+                     .topic(topicName)
+                     .subscriptionName(SUBSCRIPTION)
+                     .subscriptionType(SubscriptionType.Shared)
+                     .consumerName("departing")
+                     .receiverQueueSize(5)
+                     .subscribe()) {
+            for (int i = 0; i < UNACKED_MESSAGES; i++) {
+                producer.send("unacked-" + i);
+            }
+            for (int i = 0; i < UNACKED_MESSAGES; i++) {
+                assertThat(departing.receive(2, TimeUnit.SECONDS))
+                        .as("delivery %s to leave unacknowledged", i).isNotNull();
+            }
+
+            BrokerService brokerService = pulsar.getBrokerService();
+            PersistentTopic topic = (PersistentTopic) brokerService.getTopicReference(topicName).orElseThrow();
+            AbstractPersistentDispatcherMultipleConsumers dispatcher =
+                    (AbstractPersistentDispatcherMultipleConsumers) topic.getSubscription(SUBSCRIPTION).getDispatcher();
+            assertThat(dispatcher).as("configured dispatcher implementation").isInstanceOf(classic
+                    ? PersistentDispatcherMultipleConsumersClassic.class : PersistentDispatcherMultipleConsumers.class);
+            Consumer brokerConsumer = dispatcher.getConsumers().get(0);
+
+            // Serialize with dispatch so both aggregate credits have completed before checking their values.
+            synchronized (dispatcher) {
+                assertThat(brokerConsumer.getUnackedMessages()).as("departing consumer balance")
+                        .isEqualTo(UNACKED_MESSAGES);
+                assertThat(dispatcher.getTotalUnackedMessages()).as("subscription balance before removal")
+                        .isEqualTo(UNACKED_MESSAGES);
+                assertThat(brokerService.getTotalUnackedMessages()).as("broker balance before removal")
+                        .isEqualTo(UNACKED_MESSAGES);
+
+                // With no survivor there is no replay to race with these assertions. Checking the broker as well
+                // also rejects a fix that merely resets the subscription counter when the last consumer leaves.
+                dispatcher.removeConsumer(brokerConsumer);
+                assertUnackedMessagesCleared(dispatcher, brokerService, "first removal");
+                dispatcher.removeConsumer(brokerConsumer);
+                assertUnackedMessagesCleared(dispatcher, brokerService, "repeated removal");
+            }
+        }
+    }
+
+    private void assertUnackedMessagesCleared(AbstractPersistentDispatcherMultipleConsumers dispatcher,
+                                              BrokerService brokerService, String removal) {
+        assertThat(dispatcher.getTotalUnackedMessages()).as("subscription balance after %s", removal).isZero();
+        assertThat(brokerService.getTotalUnackedMessages()).as("broker balance after %s", removal).isZero();
+    }
+}