You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/09/10 10:24:46 UTC
(pulsar) branch master updated: [fix][broker] Fix persistent throughput degradation caused by permit loss during frequent reconnects on Shared subscriptions (#26289)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 12b6497eefd7 [fix][broker] Fix persistent throughput degradation caused by permit loss during frequent reconnects on Shared subscriptions (#26289)
12b6497eefd7 is described below
commit 12b6497eefd7829513de012dd76d77f00af803bf
Author: void-ptr974 <vo...@protonmail.com>
AuthorDate: Thu Sep 10 18:24:39 2026 +0800
[fix][broker] Fix persistent throughput degradation caused by permit loss during frequent reconnects on Shared subscriptions (#26289)
---
.../AbstractDispatcherMultipleConsumers.java | 15 +-
.../org/apache/pulsar/broker/service/Consumer.java | 67 ++-
.../PersistentDispatcherMultipleConsumers.java | 27 +-
...rsistentDispatcherMultipleConsumersClassic.java | 27 +-
.../SharedDispatcherPermitAccountingTest.java | 522 +++++++++++++++++++++
5 files changed, 643 insertions(+), 15 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherMultipleConsumers.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherMultipleConsumers.java
index bec02e94c79a..5d1dd116e85b 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherMultipleConsumers.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherMultipleConsumers.java
@@ -32,7 +32,8 @@ import org.apache.pulsar.common.api.proto.CommandSubscribe.SubType;
public abstract class AbstractDispatcherMultipleConsumers extends AbstractBaseDispatcher {
protected final CopyOnWriteArrayList<Consumer> consumerList = new CopyOnWriteArrayList<>();
- protected final ObjectSet<Consumer> consumerSet = new ObjectHashSet<>();
+ private final ObjectHashSet<Consumer> consumerSetImpl = new ObjectHashSet<>();
+ protected final ObjectSet<Consumer> consumerSet = consumerSetImpl;
protected volatile int currentConsumerRoundRobinIndex = 0;
protected static final int FALSE = 0;
@@ -58,6 +59,18 @@ public abstract class AbstractDispatcherMultipleConsumers extends AbstractBaseDi
return consumerList.size() == 1 && consumerSet.contains(consumer);
}
+ /**
+ * Checks whether the exact Consumer instance is still connected.
+ *
+ * <p>This differs from {@link ObjectSet#contains(Object)}, which uses {@link Consumer#equals(Object)} and can
+ * match a replacement Consumer that reuses the same protocol identity.
+ * The caller must hold the dispatcher monitor while checking membership and acting on the result.
+ */
+ protected final boolean containsConsumerInstance(Consumer consumer) {
+ int index = consumerSetImpl.indexOf(consumer);
+ return consumerSetImpl.indexExists(index) && consumerSetImpl.indexGet(index) == consumer;
+ }
+
public boolean isClosed() {
return isClosed == TRUE;
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java
index a7f296536881..b45947b76e83 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java
@@ -117,6 +117,18 @@ public class Consumer {
private static final AtomicIntegerFieldUpdater<Consumer> MESSAGE_PERMITS_UPDATER =
AtomicIntegerFieldUpdater.newUpdater(Consumer.class, "messagePermits");
private volatile int messagePermits = 0;
+ /**
+ * Guards the Flow-side compound update of {@link #messagePermits} and
+ * {@link #pendingDispatcherFlowPermits}. A Flow command increases the consumer permits before the dispatcher
+ * processes the corresponding update asynchronously. Consumer removal can happen between those two operations,
+ * so both values must be observed consistently when calculating how many permits are already included in the
+ * dispatcher total.
+ *
+ * <p>The dispatcher callback is invoked only after this lock is released. This avoids holding the lock while
+ * calling into the subscription and preserves the lock order used by dispatcher flow processing and removal.
+ */
+ private final Object flowPermitAccountingLock = new Object();
+ private int pendingDispatcherFlowPermits = 0;
// It starts keep tracking of messagePermits once consumer gets blocked, as consumer needs two separate counts:
// messagePermits (1) before and (2) after being blocked: to dispatch only blockedPermit number of messages at the
// time of redelivery
@@ -916,7 +928,7 @@ public class Consumer {
}
int oldPermits;
if (!blockedConsumerOnUnackedMsgs) {
- oldPermits = MESSAGE_PERMITS_UPDATER.getAndAdd(this, additionalNumberOfMessages);
+ oldPermits = addPermitsPendingDispatcherUpdate(additionalNumberOfMessages);
log.debug()
.attr("additionalNumberOfMessages", additionalNumberOfMessages)
.log("Added message permits before updating dispatcher");
@@ -943,7 +955,7 @@ public class Consumer {
void flowConsumerBlockedPermits(Consumer consumer) {
int additionalNumberOfPermits = PERMITS_RECEIVED_WHILE_CONSUMER_BLOCKED_UPDATER.getAndSet(consumer, 0);
// add newly flow permits to actual consumer.messagePermits
- MESSAGE_PERMITS_UPDATER.getAndAdd(consumer, additionalNumberOfPermits);
+ consumer.addPermitsPendingDispatcherUpdate(additionalNumberOfPermits);
log.debug()
.attr("additionalNumberOfPermits", additionalNumberOfPermits)
.log("Added blocked permits");
@@ -955,6 +967,57 @@ public class Consumer {
return MESSAGE_PERMITS_UPDATER.get(this);
}
+ /**
+ * Adds permits after a Flow command is accepted and immediately before notifying the dispatcher. The pending
+ * count covers the interval until the dispatcher's asynchronous Flow task starts processing the same permits.
+ */
+ private int addPermitsPendingDispatcherUpdate(int additionalNumberOfPermits) {
+ if (!shouldTrackPendingDispatcherFlowPermits()) {
+ return MESSAGE_PERMITS_UPDATER.getAndAdd(this, additionalNumberOfPermits);
+ }
+ synchronized (flowPermitAccountingLock) {
+ pendingDispatcherFlowPermits += additionalNumberOfPermits;
+ return MESSAGE_PERMITS_UPDATER.getAndAdd(this, additionalNumberOfPermits);
+ }
+ }
+
+ private boolean shouldTrackPendingDispatcherFlowPermits() {
+ return isPersistentTopic && Subscription.isIndividualAckMode(subType);
+ }
+
+ /**
+ * Called at the start of the dispatcher's asynchronous Flow task, before checking whether this consumer is still
+ * connected. At this point the Flow update is no longer pending: the dispatcher will either add the permits to
+ * its total or ignore them because the consumer has already been removed.
+ */
+ public void completePendingDispatcherFlow(int additionalNumberOfPermits) {
+ synchronized (flowPermitAccountingLock) {
+ // Preserve the accounting delta across signed int wrap, matching the other permit counters.
+ pendingDispatcherFlowPermits -= additionalNumberOfPermits;
+ }
+ }
+
+ /**
+ * Called while the dispatcher removes this consumer. Permits belonging to Flow tasks that have not started yet
+ * are excluded because those permits have not been added to the dispatcher total and must not be subtracted from
+ * it during removal.
+ *
+ * <p>This accounting is enabled for persistent Shared and Key_Shared dispatchers. It relies on every dispatcher
+ * Flow task calling {@link #completePendingDispatcherFlow(int)} before applying or ignoring the update. For these
+ * dispatchers, when observed under the dispatcher monitor, the total available permits equal the sum of this
+ * balance over all connected consumers.
+ *
+ * <p>The returned balance can be negative. A pending Flow makes permits visible on the consumer before its
+ * asynchronous dispatcher update runs, so the dispatcher can consume those permits while they are still counted
+ * as pending. Subtracting the negative balance during removal is required to restore the dispatcher total; callers
+ * must not clamp it to zero.
+ */
+ public int getAvailablePermitsForDispatcherRemoval() {
+ synchronized (flowPermitAccountingLock) {
+ return MESSAGE_PERMITS_UPDATER.get(this) - pendingDispatcherFlowPermits;
+ }
+ }
+
/**
* return 0 if there is no entry dispatched yet.
*/
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 b659f6e2200d..b99a2f32810c 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
@@ -24,6 +24,7 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Lists;
import com.google.common.collect.Range;
import io.github.merlimat.slog.Logger;
+import io.netty.channel.EventLoopGroup;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collections;
@@ -36,6 +37,7 @@ import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
@@ -258,9 +260,12 @@ public class PersistentDispatcherMultipleConsumers extends AbstractPersistentDis
notifyAddedToReplay.setTrue();
}
});
- totalAvailablePermits -= consumer.getAvailablePermits();
+ // Restore the invariant that the dispatcher total equals the sum of the removal balances of the
+ // remaining consumers. Exclude Flow permits that have not updated the dispatcher total yet.
+ int availablePermits = consumer.getAvailablePermitsForDispatcherRemoval();
+ totalAvailablePermits -= availablePermits;
log.debug()
- .attr("diffAvailablePermits", consumer.getAvailablePermits())
+ .attr("availablePermits", availablePermits)
.attr("totalAvailablePermits", totalAvailablePermits)
.log("Decreased totalAvailablePermits");
if (notifyAddedToReplay.booleanValue()) {
@@ -300,13 +305,23 @@ public class PersistentDispatcherMultipleConsumers extends AbstractPersistentDis
@Override
public void consumerFlow(Consumer consumer, int additionalNumberOfMessages) {
- topic.getBrokerService().executor().execute(() -> {
- internalConsumerFlow(consumer, additionalNumberOfMessages);
- });
+ EventLoopGroup flowExecutor = topic.getBrokerService().executor();
+ try {
+ flowExecutor.execute(() -> internalConsumerFlow(consumer, additionalNumberOfMessages));
+ } catch (RejectedExecutionException e) {
+ // Leave the permits pending so removal excludes this unapplied Flow during broker shutdown.
+ log.debug()
+ .attr("consumer", consumer)
+ .attr("executorShutdown", flowExecutor.isShuttingDown())
+ .exception(e)
+ .log("Unable to schedule flow control update");
+ }
}
private synchronized void internalConsumerFlow(Consumer consumer, int additionalNumberOfMessages) {
- if (!consumerSet.contains(consumer)) {
+ // The queued Flow task is no longer pending, even if the consumer was removed while the task was waiting.
+ consumer.completePendingDispatcherFlow(additionalNumberOfMessages);
+ if (!containsConsumerInstance(consumer)) {
log.debug()
.attr("consumer", consumer)
.log("Ignoring flow control from disconnected consumer");
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 3de50042b592..64eea24744dd 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
@@ -23,6 +23,7 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Lists;
import com.google.common.collect.Range;
import io.github.merlimat.slog.Logger;
+import io.netty.channel.EventLoopGroup;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collections;
@@ -35,6 +36,7 @@ import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
@@ -244,9 +246,12 @@ public class PersistentDispatcherMultipleConsumersClassic extends AbstractPersis
consumer.getPendingAcks().forEach((ledgerId, entryId, batchSize, stickyKeyHash) -> {
addMessageToReplay(ledgerId, entryId, stickyKeyHash);
});
- totalAvailablePermits -= consumer.getAvailablePermits();
+ // Restore the invariant that the dispatcher total equals the sum of the removal balances of the
+ // remaining consumers. Exclude Flow permits that have not updated the dispatcher total yet.
+ int availablePermits = consumer.getAvailablePermitsForDispatcherRemoval();
+ totalAvailablePermits -= availablePermits;
log.debug()
- .attr("availablePermits", consumer.getAvailablePermits())
+ .attr("availablePermits", availablePermits)
.attr("totalAvailablePermits", totalAvailablePermits)
.log("Decreased totalAvailablePermits by in PersistentDispatcherMultipleConsumers. "
+ "New dispatcher permit count is");
@@ -280,13 +285,23 @@ public class PersistentDispatcherMultipleConsumersClassic extends AbstractPersis
@Override
public void consumerFlow(Consumer consumer, int additionalNumberOfMessages) {
- topic.getBrokerService().executor().execute(() -> {
- internalConsumerFlow(consumer, additionalNumberOfMessages);
- });
+ EventLoopGroup flowExecutor = topic.getBrokerService().executor();
+ try {
+ flowExecutor.execute(() -> internalConsumerFlow(consumer, additionalNumberOfMessages));
+ } catch (RejectedExecutionException e) {
+ // Leave the permits pending so removal excludes this unapplied Flow during broker shutdown.
+ log.debug()
+ .attr("consumer", consumer)
+ .attr("executorShutdown", flowExecutor.isShuttingDown())
+ .exception(e)
+ .log("Unable to schedule flow control update");
+ }
}
private synchronized void internalConsumerFlow(Consumer consumer, int additionalNumberOfMessages) {
- if (!consumerSet.contains(consumer)) {
+ // The queued Flow task is no longer pending, even if the consumer was removed while the task was waiting.
+ consumer.completePendingDispatcherFlow(additionalNumberOfMessages);
+ if (!containsConsumerInstance(consumer)) {
log.debug()
.attr("consumer", consumer)
.log("Ignoring flow control from disconnected consumer");
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/SharedDispatcherPermitAccountingTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/SharedDispatcherPermitAccountingTest.java
new file mode 100644
index 000000000000..36f7bf900112
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/SharedDispatcherPermitAccountingTest.java
@@ -0,0 +1,522 @@
+/*
+ * 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 java.util.Collections.emptyMap;
+import static org.apache.pulsar.common.api.proto.CommandSubscribe.SubType.Exclusive;
+import static org.apache.pulsar.common.api.proto.CommandSubscribe.SubType.Failover;
+import static org.apache.pulsar.common.api.proto.CommandSubscribe.SubType.Key_Shared;
+import static org.apache.pulsar.common.api.proto.CommandSubscribe.SubType.Shared;
+import static org.apache.pulsar.common.api.proto.KeySharedMode.AUTO_SPLIT;
+import static org.apache.pulsar.common.protocol.Commands.DEFAULT_CONSUMER_EPOCH;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import io.netty.channel.EventLoopGroup;
+import io.netty.util.concurrent.EventExecutor;
+import io.netty.util.concurrent.ImmediateEventExecutor;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.TimeUnit;
+import org.apache.bookkeeper.mledger.Entry;
+import org.apache.bookkeeper.mledger.ManagedCursor;
+import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
+import org.apache.pulsar.broker.service.BrokerService;
+import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.Dispatcher;
+import org.apache.pulsar.broker.service.EntryBatchIndexesAcks;
+import org.apache.pulsar.broker.service.EntryBatchSizes;
+import org.apache.pulsar.broker.service.PulsarCommandSender;
+import org.apache.pulsar.broker.service.RedeliveryTracker;
+import org.apache.pulsar.broker.service.ServerCnx;
+import org.apache.pulsar.broker.service.SharedPulsarBaseTest;
+import org.apache.pulsar.broker.service.StickyKeyDispatcher;
+import org.apache.pulsar.broker.service.Subscription;
+import org.apache.pulsar.broker.service.Topic;
+import org.apache.pulsar.broker.service.TransportCnx;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.Range;
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.common.api.proto.CommandSubscribe.SubType;
+import org.apache.pulsar.common.api.proto.KeySharedMeta;
+import org.apache.pulsar.common.policies.data.stats.ConsumerStatsImpl;
+import org.awaitility.Awaitility;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+@Test(groups = "broker-api")
+public class SharedDispatcherPermitAccountingTest extends SharedPulsarBaseTest {
+
+ @DataProvider(name = "dispatcherImplementations")
+ public Object[][] dispatcherImplementations() {
+ return new Object[][] {{false}, {true}};
+ }
+
+ @DataProvider(name = "flowRaceDispatcherVariants")
+ public Object[][] flowRaceDispatcherVariants() {
+ return new Object[][] {
+ {false, Shared},
+ {true, Shared},
+ {false, Key_Shared},
+ {true, Key_Shared}
+ };
+ }
+
+ @DataProvider(name = "untrackedSubscriptionVariants")
+ public Object[][] untrackedSubscriptionVariants() {
+ return new Object[][] {
+ {true, Exclusive},
+ {true, Failover},
+ {false, Shared},
+ {false, Key_Shared}
+ };
+ }
+
+ @Test(dataProvider = "untrackedSubscriptionVariants", timeOut = 30_000)
+ public void testUnrelatedSubscriptionDoesNotAccumulatePendingDispatcherPermits(
+ boolean persistent, SubType subType) throws Exception {
+ String topicName = newTopicName();
+ String subscriptionName = "sub";
+ admin.topics().createNonPartitionedTopic(topicName);
+ PersistentTopic persistentTopic = (PersistentTopic) getTopic(topicName, false).join().orElseThrow();
+ Topic consumerTopic = persistent ? persistentTopic : mock(Topic.class);
+ if (!persistent) {
+ when(consumerTopic.getBrokerService()).thenReturn(persistentTopic.getBrokerService());
+ when(consumerTopic.getHierarchyTopicPolicies())
+ .thenReturn(persistentTopic.getHierarchyTopicPolicies());
+ }
+ Subscription subscription = mock(Subscription.class);
+ when(subscription.getName()).thenReturn(subscriptionName);
+ when(subscription.getTopic()).thenReturn(consumerTopic);
+ Consumer consumer = createConsumer(subscription, subType, topicName, 1);
+
+ consumer.flowPermits(100);
+
+ assertThat(consumer.getAvailablePermits()).isEqualTo(100);
+ assertThat(consumer.getAvailablePermitsForDispatcherRemoval()).isEqualTo(100);
+ }
+
+ @Test(dataProvider = "flowRaceDispatcherVariants", timeOut = 30_000)
+ public void testFlowCommandRaceWithConsumerRemovalDoesNotLosePermits(boolean classic, SubType subType)
+ throws Exception {
+ TestContext context = createTestContext(classic, subType);
+ Consumer remainingConsumer = context.remainingConsumer();
+ Consumer removedConsumer = context.removedConsumer();
+
+ remainingConsumer.flowPermits(10);
+ removedConsumer.flowPermits(20);
+ drainBrokerWorkerGroup(context.topic());
+ assertThat(totalAvailablePermits(context.dispatcher())).isEqualTo(30);
+
+ synchronized (context.dispatcher()) {
+ // Keep the asynchronous Flow tasks queued until removal completes.
+ removedConsumer.flowPermits(400);
+ removedConsumer.flowPermits(600);
+ assertThat(totalAvailablePermits(context.dispatcher())).isEqualTo(30);
+ context.dispatcher().removeConsumer(removedConsumer);
+ }
+ drainBrokerWorkerGroup(context.topic());
+
+ assertThat(totalAvailablePermits(context.dispatcher()))
+ .isEqualTo(remainingConsumer.getAvailablePermits());
+ }
+
+ @Test(dataProvider = "flowRaceDispatcherVariants", timeOut = 30_000)
+ public void testQueuedFlowDoesNotApplyToEqualReplacementConsumer(boolean classic, SubType subType)
+ throws Exception {
+ TestContext context = createTestContext(classic, subType);
+ Consumer original = context.removedConsumer();
+ Consumer replacement = new Consumer(original.getSubscription(), original.subType(), context.topic().getName(),
+ original.consumerId(), 0, original.consumerName(), true, original.cnx(), "role", emptyMap(), false,
+ new KeySharedMeta().setKeySharedMode(AUTO_SPLIT), MessageId.latest, DEFAULT_CONSUMER_EPOCH);
+
+ synchronized (context.dispatcher()) {
+ // Keep the queued Flow task behind the dispatcher monitor while an equal replacement joins.
+ original.flowPermits(100);
+ assertThat(original.getAvailablePermits()).isEqualTo(100);
+ assertThat(original.getAvailablePermitsForDispatcherRemoval()).isZero();
+ context.dispatcher().removeConsumer(original);
+ context.dispatcher().addConsumer(replacement).join();
+ assertThat(replacement).isNotSameAs(original).isEqualTo(original);
+ assertThat(replacement.hashCode()).isEqualTo(original.hashCode());
+ }
+ drainBrokerWorkerGroup(context.topic());
+
+ assertThat(context.dispatcher().getConsumers()).containsExactlyInAnyOrder(
+ context.remainingConsumer(), replacement);
+ assertThat(context.dispatcher().getConsumers()).anySatisfy(consumer -> assertThat(consumer)
+ .isSameAs(replacement));
+ assertThat(original.getAvailablePermitsForDispatcherRemoval()).isEqualTo(100);
+ assertThat(replacement.getAvailablePermits()).isZero();
+ assertThat(totalAvailablePermits(context.dispatcher())).isZero();
+ }
+
+ @Test(dataProvider = "dispatcherImplementations", timeOut = 30_000)
+ public void testRejectedFlowStaysPendingAndIsExcludedFromRemoval(boolean classic) throws Exception {
+ String topicName = newTopicName();
+ String subscriptionName = "shared-sub";
+ admin.topics().createNonPartitionedTopic(topicName);
+ PersistentTopic realTopic = (PersistentTopic) getTopic(topicName, false).join().orElseThrow();
+ BrokerService isolatedBroker = spy(realTopic.getBrokerService());
+ PersistentTopic isolatedTopic = spy(realTopic);
+ doReturn(isolatedBroker).when(isolatedTopic).getBrokerService();
+
+ EventLoopGroup rejectingExecutor = mock(EventLoopGroup.class);
+ doThrow(new RejectedExecutionException("test rejection"))
+ .when(rejectingExecutor).execute(any(Runnable.class));
+ when(rejectingExecutor.isShuttingDown()).thenReturn(true);
+ doReturn(rejectingExecutor).when(isolatedBroker).executor();
+
+ ManagedCursor cursor = mock(ManagedCursorImpl.class);
+ when(cursor.getName()).thenReturn(subscriptionName);
+ when(cursor.isClosed()).thenReturn(true);
+ Subscription subscription = mock(PersistentSubscription.class);
+ when(subscription.getName()).thenReturn(subscriptionName);
+ when(subscription.getTopic()).thenReturn(isolatedTopic);
+ Dispatcher dispatcher = classic
+ ? new PersistentDispatcherMultipleConsumersClassic(isolatedTopic, cursor, subscription)
+ : new PersistentDispatcherMultipleConsumers(isolatedTopic, cursor, subscription);
+ doAnswer(invocation -> {
+ dispatcher.consumerFlow(invocation.getArgument(0), invocation.getArgument(1));
+ return null;
+ }).when(subscription).consumerFlow(any(), anyInt());
+
+ Consumer remainingConsumer = createConsumer(subscription, Shared, topicName, 1);
+ Consumer removedConsumer = createConsumer(subscription, Shared, topicName, 2);
+ dispatcher.addConsumer(remainingConsumer).join();
+ dispatcher.addConsumer(removedConsumer).join();
+
+ removedConsumer.flowPermits(100);
+
+ assertThat(removedConsumer.getAvailablePermits()).isEqualTo(100);
+ assertThat(removedConsumer.getAvailablePermitsForDispatcherRemoval()).isZero();
+ assertThat(totalAvailablePermits(dispatcher)).isZero();
+ dispatcher.removeConsumer(removedConsumer);
+ assertThat(dispatcher.getConsumers()).containsExactly(remainingConsumer);
+ assertThat(removedConsumer.getAvailablePermitsForDispatcherRemoval()).isZero();
+ assertThat(totalAvailablePermits(dispatcher)).isZero();
+ verify(rejectingExecutor).execute(any(Runnable.class));
+ }
+
+ @Test(timeOut = 30_000)
+ public void testPendingPermitAccountingSurvivesSignedIntegerWrap() throws Exception {
+ String topicName = newTopicName();
+ String subscriptionName = "shared-sub";
+ admin.topics().createNonPartitionedTopic(topicName);
+ PersistentTopic topic = (PersistentTopic) getTopic(topicName, false).join().orElseThrow();
+ Subscription subscription = mock(PersistentSubscription.class);
+ when(subscription.getName()).thenReturn(subscriptionName);
+ when(subscription.getTopic()).thenReturn(topic);
+ Consumer consumer = createConsumer(subscription, Shared, topicName, 1);
+
+ consumer.flowPermits(Integer.MAX_VALUE);
+ consumer.flowPermits(Integer.MAX_VALUE);
+ consumer.flowPermits(Integer.MAX_VALUE);
+
+ assertThat(consumer.getAvailablePermits()).isEqualTo(Integer.MAX_VALUE - 2);
+ assertThat(consumer.getAvailablePermitsForDispatcherRemoval()).isZero();
+
+ consumer.completePendingDispatcherFlow(Integer.MAX_VALUE);
+ assertThat(consumer.getAvailablePermitsForDispatcherRemoval()).isEqualTo(Integer.MAX_VALUE);
+ consumer.completePendingDispatcherFlow(Integer.MAX_VALUE);
+ assertThat(consumer.getAvailablePermitsForDispatcherRemoval()).isEqualTo(-2);
+ consumer.completePendingDispatcherFlow(Integer.MAX_VALUE);
+ assertThat(consumer.getAvailablePermitsForDispatcherRemoval()).isEqualTo(Integer.MAX_VALUE - 2);
+ }
+
+ @Test(dataProvider = "flowRaceDispatcherVariants", timeOut = 30_000)
+ public void testRemovalAppliesNegativeAccountedPermitBalance(boolean classic, SubType subType) throws Exception {
+ TestContext context = createTestContext(classic, subType);
+ Consumer remainingConsumer = context.remainingConsumer();
+ Consumer removedConsumer = context.removedConsumer();
+ Consumer secondRemainingConsumer = createConsumer(
+ remainingConsumer.getSubscription(), subType, context.topic().getName(), 3);
+ context.dispatcher().addConsumer(secondRemainingConsumer).join();
+
+ remainingConsumer.flowPermits(10);
+ secondRemainingConsumer.flowPermits(15);
+ removedConsumer.flowPermits(20);
+ drainBrokerWorkerGroup(context.topic());
+ assertThat(totalAvailablePermits(context.dispatcher())).isEqualTo(45);
+
+ synchronized (context.dispatcher()) {
+ // Keep the Flow task queued, then dispatch more than the removed consumer's 20 accounted permits.
+ // Its removal balance becomes 70 available - 100 pending = -30 and must be applied as-is.
+ removedConsumer.flowPermits(100);
+ simulateDispatch(context.dispatcher(), removedConsumer, 50);
+
+ assertThat(removedConsumer.getAvailablePermits()).isEqualTo(70);
+ assertThat(removedConsumer.getAvailablePermitsForDispatcherRemoval()).isEqualTo(-30);
+ assertThat(totalAvailablePermits(context.dispatcher())).isEqualTo(-5);
+
+ context.dispatcher().removeConsumer(removedConsumer);
+ assertThat(totalAvailablePermits(context.dispatcher())).isEqualTo(25);
+ }
+ drainBrokerWorkerGroup(context.topic());
+
+ // Keep two consumers after removal so a single-consumer read floor cannot mask dispatcher permit drift.
+ assertThat(context.dispatcher().getConsumers())
+ .containsExactlyInAnyOrder(remainingConsumer, secondRemainingConsumer);
+ assertThat(totalAvailablePermits(context.dispatcher()))
+ .isEqualTo(remainingConsumer.getAvailablePermits()
+ + secondRemainingConsumer.getAvailablePermits());
+ }
+
+ @Test(dataProvider = "flowRaceDispatcherVariants", timeOut = 30_000)
+ public void testBlockedFlowCommandRaceWithConsumerRemovalDoesNotLosePermits(
+ boolean classic, SubType subType) throws Exception {
+ TestContext context = createTestContext(classic, subType);
+ Consumer remainingConsumer = context.remainingConsumer();
+ Consumer removedConsumer = context.removedConsumer();
+
+ remainingConsumer.flowPermits(10);
+ drainBrokerWorkerGroup(context.topic());
+ assertThat(totalAvailablePermits(context.dispatcher())).isEqualTo(10);
+
+ ConsumerStatsImpl blockedStats = new ConsumerStatsImpl();
+ blockedStats.blockedConsumerOnUnackedMsgs = true;
+ removedConsumer.updateStats(blockedStats);
+ assertThat(removedConsumer.isBlocked()).isTrue();
+ assertThat(removedConsumer.getMaxUnackedMessages()).isPositive();
+
+ removedConsumer.flowPermits(1_000);
+ assertThat(removedConsumer.getAvailablePermits()).isZero();
+
+ synchronized (context.dispatcher()) {
+ // Keep the asynchronous Flow task queued until removal completes.
+ removedConsumer.updateBlockedConsumerOnUnackedMsgs(removedConsumer);
+ assertThat(totalAvailablePermits(context.dispatcher())).isEqualTo(10);
+ context.dispatcher().removeConsumer(removedConsumer);
+ }
+ drainBrokerWorkerGroup(context.topic());
+
+ assertThat(totalAvailablePermits(context.dispatcher()))
+ .isEqualTo(remainingConsumer.getAvailablePermits());
+ }
+
+ @Test(dataProvider = "dispatcherImplementations", timeOut = 30_000)
+ public void testRemainingConsumerCanContinueAfterFlowAndCloseRace(boolean classic) throws Exception {
+ int receiverQueueSize = 10;
+ int messagesToConsume = receiverQueueSize * 2;
+ int pendingFlowPermits = 1_000;
+ String topicName = newTopicName();
+ String subscriptionName = "shared-sub";
+ boolean previousClassicSetting = getConfig().isSubscriptionSharedUseClassicPersistentImplementation();
+ getConfig().setSubscriptionSharedUseClassicPersistentImplementation(classic);
+ try {
+ admin.topics().createNonPartitionedTopic(topicName);
+
+ try (Producer<byte[]> producer = pulsarClient.newProducer()
+ .topic(topicName)
+ .enableBatching(false)
+ .create();
+ org.apache.pulsar.client.api.Consumer<byte[]> remainingClient = pulsarClient.newConsumer()
+ .topic(topicName)
+ .subscriptionName(subscriptionName)
+ .subscriptionType(SubscriptionType.Shared)
+ .consumerName("remaining-consumer")
+ .receiverQueueSize(receiverQueueSize)
+ .subscribe();
+ org.apache.pulsar.client.api.Consumer<byte[]> removedClient = pulsarClient.newConsumer()
+ .topic(topicName)
+ .subscriptionName(subscriptionName)
+ .subscriptionType(SubscriptionType.Shared)
+ .consumerName("removed-consumer")
+ .receiverQueueSize(receiverQueueSize)
+ .subscribe()) {
+ PersistentTopic topic = (PersistentTopic) getTopic(topicName, false).join().orElseThrow();
+ PersistentSubscription subscription =
+ (PersistentSubscription) topic.getSubscription(subscriptionName);
+ Dispatcher dispatcher = subscription.getDispatcher();
+ assertThat(dispatcher).isInstanceOf(classic
+ ? PersistentDispatcherMultipleConsumersClassic.class
+ : PersistentDispatcherMultipleConsumers.class);
+ Consumer remainingBrokerConsumer = findConsumer(dispatcher, "remaining-consumer");
+ Consumer removedBrokerConsumer = findConsumer(dispatcher, "removed-consumer");
+ ServerCnx removedConsumerCnx = (ServerCnx) removedBrokerConsumer.cnx();
+
+ Awaitility.await().atMost(Duration.ofSeconds(5)).untilAsserted(() -> {
+ assertThat(remainingBrokerConsumer.getAvailablePermits()).isEqualTo(receiverQueueSize);
+ assertThat(removedBrokerConsumer.getAvailablePermits()).isEqualTo(receiverQueueSize);
+ });
+ drainBrokerWorkerGroup(topic);
+ assertThat(totalAvailablePermits(dispatcher)).isEqualTo(receiverQueueSize * 2);
+
+ // Follow the production subscription -> dispatcher lock order and hold the dispatcher monitor so the
+ // asynchronous Flow task cannot run before Consumer.close removes the consumer.
+ synchronized (subscription) {
+ synchronized (dispatcher) {
+ removedBrokerConsumer.flowPermits(pendingFlowPermits);
+ assertThat(removedBrokerConsumer.getAvailablePermits())
+ .isEqualTo(receiverQueueSize + pendingFlowPermits);
+ assertThat(totalAvailablePermits(dispatcher)).isEqualTo(receiverQueueSize * 2);
+ removedBrokerConsumer.close();
+ }
+ }
+ // Wait for the broker-side close to remove the consumer from the connection map. The client close is
+ // then handled idempotently instead of attempting a second dispatcher removal.
+ Awaitility.await().atMost(Duration.ofSeconds(5)).untilAsserted(() ->
+ assertThat(removedConsumerCnx.getConsumers()
+ .containsKey(removedBrokerConsumer.consumerId())).isFalse());
+ removedClient.close();
+ drainBrokerWorkerGroup(topic);
+
+ assertThat(dispatcher.getConsumers()).containsExactly(remainingBrokerConsumer);
+ assertThat(totalAvailablePermits(dispatcher))
+ .isEqualTo(remainingBrokerConsumer.getAvailablePermits());
+
+ for (int i = 0; i < messagesToConsume; i++) {
+ producer.send(new byte[] {(byte) i});
+ }
+ for (int i = 0; i < messagesToConsume; i++) {
+ Message<byte[]> message = remainingClient.receive(5, TimeUnit.SECONDS);
+ assertThat(message).isNotNull();
+ remainingClient.acknowledge(message);
+ }
+ }
+ } finally {
+ getConfig().setSubscriptionSharedUseClassicPersistentImplementation(previousClassicSetting);
+ }
+ }
+
+ private TestContext createTestContext(boolean classic) throws Exception {
+ return createTestContext(classic, Shared);
+ }
+
+ private TestContext createTestContext(boolean classic, SubType subType) throws Exception {
+ String topicName = newTopicName();
+ String subscriptionName = "shared-sub";
+ admin.topics().createNonPartitionedTopic(topicName);
+
+ PersistentTopic topic = (PersistentTopic) getTopic(topicName, false).join().get();
+ ManagedCursor cursor = mock(ManagedCursorImpl.class);
+ when(cursor.getName()).thenReturn(subscriptionName);
+ when(cursor.isClosed()).thenReturn(true);
+ Subscription subscription = mock(PersistentSubscription.class);
+ when(subscription.getName()).thenReturn(subscriptionName);
+ when(subscription.getTopic()).thenReturn(topic);
+
+ Dispatcher dispatcher;
+ if (subType == Key_Shared) {
+ dispatcher = classic
+ ? new PersistentStickyKeyDispatcherMultipleConsumersClassic(
+ topic, cursor, subscription, getConfig(), new KeySharedMeta().setKeySharedMode(AUTO_SPLIT))
+ : new PersistentStickyKeyDispatcherMultipleConsumers(
+ topic, cursor, subscription, getConfig(), new KeySharedMeta().setKeySharedMode(AUTO_SPLIT));
+ } else {
+ dispatcher = classic
+ ? new PersistentDispatcherMultipleConsumersClassic(topic, cursor, subscription)
+ : new PersistentDispatcherMultipleConsumers(topic, cursor, subscription);
+ }
+ doAnswer(invocation -> {
+ dispatcher.consumerFlow(invocation.getArgument(0), invocation.getArgument(1));
+ return null;
+ }).when(subscription).consumerFlow(any(), anyInt());
+
+ Consumer remainingConsumer = createConsumer(subscription, subType, topicName, 1);
+ Consumer removedConsumer = createConsumer(subscription, subType, topicName, 2);
+ dispatcher.addConsumer(remainingConsumer).join();
+ dispatcher.addConsumer(removedConsumer).join();
+
+ return new TestContext(topic, dispatcher, remainingConsumer, removedConsumer);
+ }
+
+ private Consumer createConsumer(Subscription subscription, SubType subType, String topicName, long consumerId) {
+ TransportCnx cnx = mock(TransportCnx.class);
+ PulsarCommandSender commandSender = mock(PulsarCommandSender.class);
+ when(cnx.isActive()).thenReturn(true);
+ when(cnx.isWritable()).thenReturn(true);
+ when(cnx.getCommandSender()).thenReturn(commandSender);
+ when(commandSender.sendMessagesToConsumer(anyLong(), anyString(), any(), anyInt(), anyList(),
+ any(EntryBatchSizes.class), any(EntryBatchIndexesAcks.class), any(RedeliveryTracker.class), anyLong()))
+ .thenReturn(ImmediateEventExecutor.INSTANCE.newSucceededFuture(null));
+ return new Consumer(subscription, subType, topicName, consumerId, 0, "consumer-" + consumerId,
+ true, cnx, "role", emptyMap(), false, new KeySharedMeta().setKeySharedMode(AUTO_SPLIT),
+ MessageId.latest, DEFAULT_CONSUMER_EPOCH);
+ }
+
+ private static void simulateDispatch(Dispatcher dispatcher, Consumer consumer, int permits) {
+ Entry entry = mock(Entry.class);
+ when(entry.getLedgerId()).thenReturn(1L);
+ when(entry.getEntryId()).thenReturn(1L);
+ EntryBatchSizes batchSizes = EntryBatchSizes.get(1);
+ batchSizes.setBatchSize(0, permits);
+ EntryBatchIndexesAcks batchIndexesAcks = EntryBatchIndexesAcks.get(1);
+ RedeliveryTracker redeliveryTracker = mock(RedeliveryTracker.class);
+ List<Integer> stickyKeyHashes = null;
+ if (dispatcher instanceof StickyKeyDispatcher stickyKeyDispatcher) {
+ List<Range> ranges = stickyKeyDispatcher.getConsumerKeyHashRanges().get(consumer);
+ assertThat(ranges).isNotEmpty();
+ stickyKeyHashes = List.of(ranges.get(0).getStart());
+ }
+
+ consumer.sendMessages(new ArrayList<>(List.of(entry)), stickyKeyHashes, batchSizes, batchIndexesAcks,
+ permits, 0, 0, redeliveryTracker, DEFAULT_CONSUMER_EPOCH).syncUninterruptibly();
+ decrementTotalAvailablePermits(dispatcher, permits);
+
+ batchSizes.recyle();
+ batchIndexesAcks.recycle();
+ }
+
+ private static void decrementTotalAvailablePermits(Dispatcher dispatcher, int permits) {
+ if (dispatcher instanceof PersistentDispatcherMultipleConsumers pip379Dispatcher) {
+ pip379Dispatcher.totalAvailablePermits -= permits;
+ } else {
+ ((PersistentDispatcherMultipleConsumersClassic) dispatcher).totalAvailablePermits -= permits;
+ }
+ }
+
+ private static void drainBrokerWorkerGroup(PersistentTopic topic) throws Exception {
+ for (EventExecutor eventExecutor : topic.getBrokerService().executor()) {
+ eventExecutor.submit(() -> { }).sync();
+ }
+ }
+
+ private static int totalAvailablePermits(Dispatcher dispatcher) {
+ if (dispatcher instanceof PersistentDispatcherMultipleConsumers pip379Dispatcher) {
+ return pip379Dispatcher.totalAvailablePermits;
+ }
+ return ((PersistentDispatcherMultipleConsumersClassic) dispatcher).totalAvailablePermits;
+ }
+
+ private static Consumer findConsumer(Dispatcher dispatcher, String consumerName) {
+ return dispatcher.getConsumers().stream()
+ .filter(consumer -> consumerName.equals(consumer.consumerName()))
+ .findFirst()
+ .orElseThrow();
+ }
+
+ private record TestContext(PersistentTopic topic, Dispatcher dispatcher,
+ Consumer remainingConsumer, Consumer removedConsumer) {
+ }
+}