From fbf8ca7b0b36196579fe43146beb4d0ab7b01f37 Mon Sep 17 00:00:00 2001 From: Florentin Dubois Date: Tue, 25 Aug 2026 20:41:59 +0000 Subject: [PATCH] [fix][broker] Debit un-acked messages only when the consumer is actually removed MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ### Motivation `PersistentDispatcherMultipleConsumers#removeConsumer` debits the subscription's un-acknowledged message count by the departing consumer's own count before it establishes whether that consumer is still registered: addUnAckedMessages(-consumer.getUnackedMessages()); if (consumerSet.removeAll(consumer) == 1) { The `else` branch below is the defensive path added by #22270 for a consumer that is still in `consumerList` but no longer in `consumerSet`, so that the topic can still be unloaded. Reaching it means the consumer was already removed once, and the first removal already debited its un-acknowledged messages. The unguarded debit therefore subtracts them a second time and drives `totalUnackedMessages` negative. That counter is what `maxUnackedMessagesOnSubscription` throttles on, and nothing resets it while the dispatcher lives — `clearComponentsAfterRemovedAllConsumers()` resets the available-permits aggregate but deliberately leaves it alone. A negative value therefore silently disables the throttle for the lifetime of the dispatcher, and since `addUnAckedMessages` also feeds the broker-wide counter, the drift is not confined to one subscription. `PersistentDispatcherMultipleConsumersClassic#removeConsumer` carries the identical unguarded debit. ### Modifications Move the debit inside the `consumerSet.removeAll(consumer) == 1` guard in both the current and the classic dispatcher, so that only the removal which actually unregisters the consumer accounts for it. The defensive branch needs no debit of its own, for the same reason: the first removal already made it — a comment in that branch now records the invariant. ### Verifying this change Adds `SharedSubscriptionUnackedMessagesAccountingTest`, which leaves a consumer holding ten un-acknowledged deliveries, removes it twice and requires the subscription counter to end at zero — once against the current dispatcher and once against the classic one behind the dynamic `subscriptionSharedUseClassicPersistentImplementation` flag. Without this change both end at -10. This is broker-internal accounting: no public API, configuration or wire-protocol change. Signed-off-by: Florentin Dubois --- ...PersistentDispatcherMultipleConsumers.java | 8 +- ...entDispatcherMultipleConsumersClassic.java | 8 +- ...criptionUnackedMessagesAccountingTest.java | 221 ++++++++++++++++++ 3 files changed, 233 insertions(+), 4 deletions(-) create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/SharedSubscriptionUnackedMessagesAccountingTest.java 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 b659f6e2200d8..5ba07cce6519c 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 @@ -239,9 +239,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) @@ -274,6 +276,8 @@ 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"); + // No un-acked debit here: reaching this branch means the consumer already left + // consumerSet, so the removal that unregistered it has debited its messages. 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 3de50042b592d..e869910b1debe 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 @@ -229,9 +229,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) @@ -259,6 +261,8 @@ 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"); + // No un-acked debit here: reaching this branch means the consumer already left + // consumerSet, so the removal that unregistered it has debited its messages. consumerList.removeIf(c -> consumer.equals(c)); if (consumerList.isEmpty()) { clearComponentsAfterRemovedAllConsumers(); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/SharedSubscriptionUnackedMessagesAccountingTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/SharedSubscriptionUnackedMessagesAccountingTest.java new file mode 100644 index 0000000000000..5aaa907149f4e --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/SharedSubscriptionUnackedMessagesAccountingTest.java @@ -0,0 +1,221 @@ +/* + * 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.persistent; + +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertFalse; +import static org.testng.Assert.assertNotNull; +import static org.testng.Assert.assertTrue; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.concurrent.TimeUnit; +import org.apache.pulsar.broker.service.SharedPulsarBaseTest; +import org.apache.pulsar.client.api.Consumer; +import org.apache.pulsar.client.api.Producer; +import org.apache.pulsar.client.api.PulsarClient; +import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.client.api.SubscriptionType; +import org.awaitility.Awaitility; +import org.testng.annotations.Test; + +/** + * Guards the un-acknowledged message accounting of a Shared subscription on the consumer-removal + * path, for both the current ({@link PersistentDispatcherMultipleConsumers}) and the classic + * ({@link PersistentDispatcherMultipleConsumersClassic}) dispatcher implementations. + * + *

{@code removeConsumer} must debit the subscription's un-acknowledged message count exactly + * once per consumer, on the removal that actually unregisters it. That counter is what + * {@code maxUnackedMessagesOnSubscription} throttles on, it feeds the broker-wide counter through + * {@code addUnAckedMessages}, and nothing resets it while the dispatcher lives — + * {@code clearComponentsAfterRemovedAllConsumers()} resets the available-permits aggregate but + * deliberately leaves it alone — so a double debit silently raises the effective limit for the + * lifetime of the dispatcher. + */ +@Test(groups = "broker-api") +public class SharedSubscriptionUnackedMessagesAccountingTest extends SharedPulsarBaseTest { + + private static final String SUBSCRIPTION = "shared-churn-sub"; + private static final int RECEIVER_QUEUE_SIZE = 5; + private static final int UNACKED_MESSAGES = 10; + + private static final String CLASSIC_DISPATCHER_FLAG = "subscriptionSharedUseClassicPersistentImplementation"; + + /** + * Deterministic probe for the double-debit of the subscription's un-acknowledged message count + * on the consumer-removal path. + * + *

{@code PersistentDispatcherMultipleConsumers#removeConsumer(Consumer)} debits the + * subscription by the departing consumer's un-acknowledged message count before it establishes + * whether that consumer was still registered at all. Removing the same consumer twice — which + * the defensive path of apache/pulsar#22270 + * exists precisely to tolerate — therefore debits the same deliveries twice and drives the + * subscription counter negative. That counter is what + * {@code maxUnackedMessagesOnSubscription} throttles on, so a negative value silently disables + * the throttle for the lifetime of the dispatcher. + * + *

A single consumer is attached on purpose: with a second consumer connected, the first + * removal replays the departing consumer's pending acknowledgements to the survivor, which + * credits the counter again on a timing the test cannot observe. Removing the only consumer + * takes {@code clearComponentsAfterRemovedAllConsumers()}, which resets the available-permits + * aggregate but deliberately leaves the un-acknowledged count alone, so the double debit stays + * observable. + */ + @Test(timeOut = 60_000) + public void testRemovingSameConsumerTwiceDebitsUnackedMessagesOnce() throws Exception { + final String topicName = newTopicName(); + admin.topics().createNonPartitionedTopic(topicName); + + try (PulsarClient departingClient = newPulsarClient(); + Producer producer = pulsarClient.newProducer() + .topic(topicName) + .enableBatching(false) + .create()) { + Consumer departing = departingClient.newConsumer(Schema.BYTES) + .topic(topicName) + .subscriptionName(SUBSCRIPTION) + .subscriptionType(SubscriptionType.Shared) + .consumerName("departing") + .receiverQueueSize(RECEIVER_QUEUE_SIZE) + .subscribe(); + + for (int i = 0; i < UNACKED_MESSAGES; i++) { + producer.send(("unacked-" + i).getBytes(StandardCharsets.UTF_8)); + } + for (int i = 0; i < UNACKED_MESSAGES; i++) { + assertNotNull(departing.receive(30, TimeUnit.SECONDS), + "the consumer did not receive the delivery it has to leave un-acknowledged"); + } + + PersistentDispatcherMultipleConsumers dispatcher = sharedDispatcher(topicName); + org.apache.pulsar.broker.service.Consumer brokerConsumer = + brokerConsumer(dispatcher, "departing"); + Awaitility.await().atMost(Duration.ofSeconds(30)).untilAsserted(() -> { + assertEquals(brokerConsumer.getUnackedMessages(), UNACKED_MESSAGES); + assertEquals(dispatcher.totalUnackedMessages, UNACKED_MESSAGES); + }); + + dispatcher.removeConsumer(brokerConsumer); + // The consumer is no longer registered, so this removal must not debit its + // un-acknowledged messages a second time. + dispatcher.removeConsumer(brokerConsumer); + + assertEquals(dispatcher.totalUnackedMessages, 0, + "removing an already-removed consumer debited its " + UNACKED_MESSAGES + + " un-acknowledged messages from the subscription a second time"); + + departing.close(); + } + } + + /** + * The classic dispatcher ({@code subscriptionSharedUseClassicPersistentImplementation=true}, + * the documented PIP-379 rollback path) carries the identical unguarded debit in its own + * {@code removeConsumer}, so the same probe is run against it. The flag is dynamic and the + * dispatcher implementation is chosen when the first consumer attaches, so it is flipped for + * the duration of this test only and restored afterwards. + */ + @Test(timeOut = 60_000) + public void testRemovingSameConsumerTwiceDebitsUnackedMessagesOnceOnClassicDispatcher() throws Exception { + admin.brokers().updateDynamicConfiguration(CLASSIC_DISPATCHER_FLAG, "true"); + try { + Awaitility.await().atMost(Duration.ofSeconds(30)).untilAsserted(() -> assertTrue( + getPulsar().getConfiguration().isSubscriptionSharedUseClassicPersistentImplementation(), + "the classic-dispatcher flag did not propagate to the broker")); + + final String topicName = newTopicName(); + admin.topics().createNonPartitionedTopic(topicName); + + try (PulsarClient departingClient = newPulsarClient(); + Producer producer = pulsarClient.newProducer() + .topic(topicName) + .enableBatching(false) + .create()) { + Consumer departing = departingClient.newConsumer(Schema.BYTES) + .topic(topicName) + .subscriptionName(SUBSCRIPTION) + .subscriptionType(SubscriptionType.Shared) + .consumerName("departing") + .receiverQueueSize(RECEIVER_QUEUE_SIZE) + .subscribe(); + + for (int i = 0; i < UNACKED_MESSAGES; i++) { + producer.send(("unacked-" + i).getBytes(StandardCharsets.UTF_8)); + } + for (int i = 0; i < UNACKED_MESSAGES; i++) { + assertNotNull(departing.receive(30, TimeUnit.SECONDS), + "the consumer did not receive the delivery it has to leave un-acknowledged"); + } + + PersistentDispatcherMultipleConsumersClassic dispatcher = classicDispatcher(topicName); + org.apache.pulsar.broker.service.Consumer brokerConsumer = + brokerConsumer(dispatcher, "departing"); + Awaitility.await().atMost(Duration.ofSeconds(30)).untilAsserted(() -> { + assertEquals(brokerConsumer.getUnackedMessages(), UNACKED_MESSAGES); + assertEquals(dispatcher.totalUnackedMessages, UNACKED_MESSAGES); + }); + + dispatcher.removeConsumer(brokerConsumer); + // The consumer is no longer registered, so this removal must not debit its + // un-acknowledged messages a second time. + dispatcher.removeConsumer(brokerConsumer); + + assertEquals(dispatcher.totalUnackedMessages, 0, + "removing an already-removed consumer debited its " + UNACKED_MESSAGES + + " un-acknowledged messages from the subscription a second time"); + + departing.close(); + } + } finally { + admin.brokers().updateDynamicConfiguration(CLASSIC_DISPATCHER_FLAG, "false"); + Awaitility.await().atMost(Duration.ofSeconds(30)).untilAsserted(() -> assertFalse( + getPulsar().getConfiguration().isSubscriptionSharedUseClassicPersistentImplementation(), + "the classic-dispatcher flag was not restored")); + } + } + + private PersistentDispatcherMultipleConsumers sharedDispatcher(String topicName) { + AbstractPersistentDispatcherMultipleConsumers dispatcher = dispatcher(topicName); + assertTrue(dispatcher instanceof PersistentDispatcherMultipleConsumers, + "expected the current dispatcher implementation, got " + dispatcher.getClass().getSimpleName()); + return (PersistentDispatcherMultipleConsumers) dispatcher; + } + + private PersistentDispatcherMultipleConsumersClassic classicDispatcher(String topicName) { + AbstractPersistentDispatcherMultipleConsumers dispatcher = dispatcher(topicName); + assertTrue(dispatcher instanceof PersistentDispatcherMultipleConsumersClassic, + "expected the classic dispatcher implementation, got " + dispatcher.getClass().getSimpleName()); + return (PersistentDispatcherMultipleConsumersClassic) dispatcher; + } + + private AbstractPersistentDispatcherMultipleConsumers dispatcher(String topicName) { + PersistentTopic topic = (PersistentTopic) getTopicIfExists(topicName).join() + .orElseThrow(() -> new IllegalStateException("topic is not loaded: " + topicName)); + PersistentSubscription subscription = topic.getSubscription(SUBSCRIPTION); + assertNotNull(subscription, "subscription is missing: " + SUBSCRIPTION); + return (AbstractPersistentDispatcherMultipleConsumers) subscription.getDispatcher(); + } + + private org.apache.pulsar.broker.service.Consumer brokerConsumer( + AbstractPersistentDispatcherMultipleConsumers dispatcher, String consumerName) { + return dispatcher.getConsumers().stream() + .filter(consumer -> consumerName.equals(consumer.consumerName())) + .findFirst() + .orElseThrow(() -> new IllegalStateException("consumer is not connected: " + consumerName)); + } +}