Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
FlorentinDUBOIS wants to merge 1 commit into
base: master
Choose a base branch
from
+233 −4
Open

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 6 additions & 2 deletions ...va/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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

@lhotari lhotari Aug 31, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[QUALITY] the new else-branch comment states an invariant that internalRemoveConsumer does not satisfy

The comment reads as an absolute invariant, and there is one path where it does not hold: PersistentStickyKeyDispatcherMultipleConsumers#addConsumer calls internalRemoveConsumer(consumer) from its exceptionally handler (PersistentStickyKeyDispatcherMultipleConsumers.java:162), and internalRemoveConsumer (PersistentDispatcherMultipleConsumers.java:288) takes the consumer out of consumerSet without debiting anything. A later removeConsumer for that consumer lands in this branch with no earlier debit — and where the old code debited unconditionally, the new code does not.

I do not think that is a live defect on this diff, and I checked rather than assumed: super.addConsumer returns an already-completed future, and every selector's addConsumer in the tree today (ConsistentHashingStickyKeyConsumerSelector, HashRangeAutoSplitStickyKeyConsumerSelector, HashRangeExclusiveStickyKeyConsumerSelector via a synchronous validateKeySharedMeta, EntryBucketConsumerSelector) completes inline, so the whole add-and-fail sequence runs under the dispatcher monitor with no dispatch window. The consumer also has no flow permits until the client's Flow arrives, which happens after the subscribe succeeds. So getUnackedMessages() is 0 there.

Still, since this is the one behavioural difference the move introduces, I would either soften the wording (say the debit belongs to the removal that unregisters the consumer, and note that internalRemoveConsumer only ever runs for a consumer that has received nothing), or make the debit idempotent by construction instead of by branch invariant — e.g. clear the consumer's counter as you debit it (UNACKED_MESSAGES_UPDATER.getAndSet(consumer, 0) behind a small Consumer accessor). The second closes the internalRemoveConsumer gap too, at the cost of a new method on Consumer; the guard as written is correct for every path I could reach, so this is your call.

Same comment applies verbatim at PersistentDispatcherMultipleConsumersClassic.java:264.

// consumerSet, so the removal that unregistered it has debited its messages.
consumerList.removeIf(c -> consumer.equals(c));
if (consumerList.isEmpty()) {
clearComponentsAfterRemovedAllConsumers();
Expand Down
8 changes: 6 additions & 2 deletions ...apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumersClassic.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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();
Expand Down
221 changes: 221 additions & 0 deletions ...che/pulsar/broker/service/persistent/SharedSubscriptionUnackedMessagesAccountingTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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.
*
* <p>{@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.
*
* <p>{@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 <a href="https://github.com/apache/pulsar/pull/22270">apache/pulsar#22270</a>
* 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.
*
* <p>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<byte[]> producer = pulsarClient.newProducer()
.topic(topicName)
.enableBatching(false)
.create()) {
Consumer<byte[]> 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);

@lhotari lhotari Aug 31, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[QUALITY] test reads the protected totalUnackedMessages field where a public accessor exists

Nit, take or leave. totalUnackedMessages is read directly here and at :118, :170, :178; both dispatchers expose public int getTotalUnackedMessages(), which keeps the test off a protected field and off the same-package requirement.

While you are in there: CODING.md:108 asks for AssertJ assertions with descriptions in preference to the TestNG ones for anything beyond the basics — e.g. assertThat(dispatcher.getTotalUnackedMessages()).as("…").isZero(). Your failure messages are already unusually good, so this is purely about matching the house style.

One thing I want to call out positively: building the scenario by calling dispatcher.removeConsumer(...) twice, rather than reaching into consumerSet/consumerList with WhiteboxImpl the way PersistentDispatcherMultipleConsumersClassicTest does, is exactly what CODING.md asks for ("No reflection into private state"). Please keep that.

});

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,

@lhotari lhotari Aug 31, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[QUALITY] both tests would stay green under a fix that still double-debits the broker-wide counter

Both tests use the only-consumer path and assert only dispatcher.totalUnackedMessages, so they pin the subscription aggregate but not the broker-wide one that addUnAckedMessages also feeds — the very reason the description gives for the drift not being confined to one subscription.

Concretely, consider the alternative "fix" of resetting totalUnackedMessages inside clearComponentsAfterRemovedAllConsumers() and leaving the unguarded debit where master has it. First removal: 10 → 0, then clearComponents resets to 0. Second removal: debit -10 → -10, then the else branch reaches clearComponents (the list is empty) and resets to 0. Both assertions read 0 and both tests pass — while BrokerService.addUnAckedMessages(this, -10) has still been called twice.

The broker-wide counter is not directly observable in these tests because maxUnackedMessagesPerBroker defaults to 0 (ServiceConfiguration:1135) and BrokerService.addUnAckedMessages only maintains its accumulator when that is > 0 (BrokerService:4094).

A cheap way to close the gap without a second broker: keep a second consumer attached so the departing consumer's removal does not go through clearComponentsAfterRemovedAllConsumers() at all, and assert the subscription counter after both removals. That makes the assertion sensitive to where the debit happens rather than to the end-state reset. Your class Javadoc already explains why you avoided a second consumer (pending-ack replay credits the counter back on a timing the test cannot observe) — an alternative is to raise maxUnackedMessagesPerBroker on the broker and assert pulsar.getBrokerService()'s counter too, but that needs its own broker, so it may fit better with whatever you decide for the classic test above.

"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");

@lhotari lhotari Aug 31, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[QUALITY] the classic-dispatcher test can leave the whole test JVM in classic-dispatcher mode

SharedPulsarBaseTest runs against a SharedPulsarCluster singleton that lives for the whole test JVM — SharedPulsarCluster:58 (private static volatile SharedPulsarCluster instance) and :72-79, where it is created once and torn down only from a JVM shutdown hook. CODING.md:121-124 says the same thing ("It shares one SharedPulsarCluster for the test-JVM lifecycle … since the runtime is shared"). Per-method isolation is a fresh namespace and nothing more, so admin.brokers().updateDynamicConfiguration(...) changes the broker for every test class that runs after this one in the same Gradle fork.

The restore is best-effort in three ways:

  • the updateDynamicConfiguration(..., "true") call is outside the try, so a failure after the broker has applied it skips the finally entirely;
  • the finally is itself an admin call plus a 30s Awaitility wait, so it can fail or hang;
  • timeOut = 60_000 is well under this method's own worst case — up to 30s for the flag to propagate, then ten receive(30, SECONDS) calls, then a 30s Awaitility assertion. If TestNG aborts the method on its timeout, the restore may never complete.

When that happens the symptom does not appear here: later shared-cluster test classes silently get classic dispatchers and fail in ways that depend on execution order.

Every other test in the repo that needs the classic dispatcher gives itself a broker instead of mutating a shared one — ConsumerStatsTest:265, KeySharedSubscriptionTest:170, KeySharedSubscriptionMaxUnackedMessagesTest:82, KeySharedSubscriptionDisabledBrokerCacheTest:93, KeySharedSubscriptionBrokerCacheTest:91, all via conf.setSubscriptionSharedUseClassicPersistentImplementation(...). This is the only SharedPulsarBaseTest subclass in pulsar-broker that calls updateDynamicConfiguration at all.

Suggestion: keep testRemovingSameConsumerTwiceDebitsUnackedMessagesOnce here on the shared cluster, and move the classic variant into its own class with its own broker (MockedPulsarServiceBaseTest/ProducerConsumerBase + doInitConf). That also removes the propagation wait, the restore, and the timeout pressure in one go.

As a smaller aside either way: assertNotNull(departing.receive(30, TimeUnit.SECONDS), ...) ten times allows 300s inside a 60s method timeout. A shorter per-receive timeout (or receive() under the method timeout) would fail faster and more legibly.

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<byte[]> producer = pulsarClient.newProducer()
.topic(topicName)
.enableBatching(false)
.create()) {
Consumer<byte[]> 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));
}
}
Loading