diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java index 1a7375cc6299b..6d6583dc52d4d 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java +++ b/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}. diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java index b99a2f32810ce..3df819558c7b4 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java +++ b/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(); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumersClassic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumersClassic.java index 64eea24744dd5..99b7996a6115f 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumersClassic.java +++ b/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(); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SharedSubscriptionUnackedMessagesAccountingTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SharedSubscriptionUnackedMessagesAccountingTest.java new file mode 100644 index 0000000000000..ad9135b0df190 --- /dev/null +++ b/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 producer = pulsarClient.newProducer(Schema.STRING) + .topic(topicName).enableBatching(false).create(); + org.apache.pulsar.client.api.Consumer 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(); + } +}