You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by mm...@apache.org on 2026/09/15 17:05:57 UTC
(pulsar) branch master updated: [improve][broker] Optimize consumer selection for shared subscriptions (#26593)
This is an automated email from the ASF dual-hosted git repository.
merlimat 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 e7cbb33d6575 [improve][broker] Optimize consumer selection for shared subscriptions (#26593)
e7cbb33d6575 is described below
commit e7cbb33d657541fd483fdb55500f7c9ebc03a4f0
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Tue Sep 15 20:05:49 2026 +0300
[improve][broker] Optimize consumer selection for shared subscriptions (#26593)
---
.../service/ConsumerPrioritySelectorBenchmark.java | 113 ++++++++++
.../AbstractDispatcherMultipleConsumers.java | 116 ++--------
.../broker/service/ConsumerPrioritySelector.java | 168 +++++++++++++++
.../NonPersistentDispatcherMultipleConsumers.java | 4 +-
...istentStickyKeyDispatcherMultipleConsumers.java | 2 +-
.../PersistentDispatcherMultipleConsumers.java | 8 +-
...rsistentDispatcherMultipleConsumersClassic.java | 6 +-
...tickyKeyDispatcherMultipleConsumersClassic.java | 2 +-
.../service/ConsumerPrioritySelectorTest.java | 233 +++++++++++++++++++++
9 files changed, 544 insertions(+), 108 deletions(-)
diff --git a/microbench/src/main/java/org/apache/pulsar/broker/service/ConsumerPrioritySelectorBenchmark.java b/microbench/src/main/java/org/apache/pulsar/broker/service/ConsumerPrioritySelectorBenchmark.java
new file mode 100644
index 000000000000..4dbd37b1c9de
--- /dev/null
+++ b/microbench/src/main/java/org/apache/pulsar/broker/service/ConsumerPrioritySelectorBenchmark.java
@@ -0,0 +1,113 @@
+/*
+ * 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 java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.TimeUnit;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Warmup;
+
+/** Isolates selection; excludes broker setup, networking, and dispatch payload processing. */
+@State(Scope.Thread)
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.NANOSECONDS)
+@Warmup(iterations = 3, time = 1)
+@Measurement(iterations = 5, time = 1)
+@Fork(2)
+public class ConsumerPrioritySelectorBenchmark {
+ @Param({"1", "16", "50", "256"})
+ public int consumerCount;
+
+ @Param({"1", "4"})
+ public int priorityLevels;
+
+ @Param({"0", "3"})
+ public int priorityOffset;
+
+ @Param({"READY", "SPARSE", "LAST_LEVEL", "BLOCKED", "FLOW"})
+ public String availability;
+
+ private List<Slot> consumers;
+ private ConsumerPrioritySelector<Slot> selector;
+ private int cursor;
+ private int invocations;
+
+ @Setup
+ public void setup() {
+ consumers = new CopyOnWriteArrayList<>();
+ for (int i = 0; i < consumerCount; i++) {
+ int level = i * Math.min(priorityLevels, consumerCount) / consumerCount;
+ Slot slot = new Slot(priorityOffset + level * 3);
+ slot.permits = switch (availability) {
+ case "READY", "FLOW" -> 32;
+ case "SPARSE" -> i % 8 == 0 ? 32 : 0;
+ case "LAST_LEVEL" -> level == Math.min(priorityLevels, consumerCount) - 1 ? 32 : 0;
+ case "BLOCKED" -> 0;
+ default -> throw new IllegalArgumentException(availability);
+ };
+ consumers.add(slot);
+ }
+ selector = new ConsumerPrioritySelector<>(consumers, c -> c.priority, c -> c.permits > 0);
+ // Exercise wraparound even when no selection succeeds to advance the cursor.
+ cursor = consumerCount / 2;
+ }
+
+ @Benchmark
+ public int select() {
+ if (cursor >= consumers.size()) {
+ cursor = 0;
+ }
+ int selected = selector.select(cursor);
+ if (selected >= 0) {
+ cursor = selected + 1;
+ }
+ if (availability.equals("FLOW")) {
+ if (selected >= 0) {
+ consumers.get(selected).permits--;
+ }
+ // Periodic FLOW-like permit replenishment lets exhausted higher priorities become ready again.
+ if (++invocations == consumerCount * 32) {
+ for (Slot consumer : consumers) {
+ consumer.permits = 32;
+ }
+ invocations = 0;
+ }
+ }
+ return selected;
+ }
+
+ private static final class Slot {
+ private final int priority;
+ private volatile int permits;
+
+ private Slot(int priority) {
+ this.priority = priority;
+ }
+ }
+}
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 5d1dd116e85b..c1358d75aeeb 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
@@ -23,6 +23,7 @@ import com.carrotsearch.hppc.ObjectSet;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+import java.util.function.Predicate;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.common.api.proto.CommandSubscribe.SubType;
@@ -31,7 +32,10 @@ import org.apache.pulsar.common.api.proto.CommandSubscribe.SubType;
*/
public abstract class AbstractDispatcherMultipleConsumers extends AbstractBaseDispatcher {
+ // Use the membership helpers below for mutations so the selector priority counts stay consistent.
protected final CopyOnWriteArrayList<Consumer> consumerList = new CopyOnWriteArrayList<>();
+ private final ConsumerPrioritySelector<Consumer> consumerPrioritySelector =
+ new ConsumerPrioritySelector<>(consumerList, Consumer::getPriorityLevel, this::isConsumerAvailable);
private final ObjectHashSet<Consumer> consumerSetImpl = new ObjectHashSet<>();
protected final ObjectSet<Consumer> consumerSet = consumerSetImpl;
protected volatile int currentConsumerRoundRobinIndex = 0;
@@ -47,6 +51,19 @@ public abstract class AbstractDispatcherMultipleConsumers extends AbstractBaseDi
super(subscription, serviceConfig);
}
+ // Keep membership and priority counts under the same monitor used for selection.
+ protected final synchronized void addConsumerToList(Consumer consumer) {
+ consumerPrioritySelector.add(consumer);
+ }
+
+ protected final synchronized void removeConsumerFromList(Consumer consumer) {
+ consumerPrioritySelector.remove(consumer);
+ }
+
+ protected final synchronized void removeConsumersFromList(Predicate<Consumer> predicate) {
+ consumerPrioritySelector.removeIf(predicate);
+ }
+
public boolean isConsumerConnected() {
return !consumerList.isEmpty();
}
@@ -125,7 +142,7 @@ public abstract class AbstractDispatcherMultipleConsumers extends AbstractBaseDi
*
* @return nextAvailableConsumer
*/
- public Consumer getNextConsumer() {
+ public synchronized Consumer getNextConsumer() {
if (consumerList.isEmpty() || IS_CLOSED_UPDATER.get(this) == TRUE) {
// abort read if no consumers are connected or if disconnect is initiated
return null;
@@ -135,26 +152,11 @@ public abstract class AbstractDispatcherMultipleConsumers extends AbstractBaseDi
currentConsumerRoundRobinIndex = 0;
}
- int currentRoundRobinConsumerPriority = consumerList.get(currentConsumerRoundRobinIndex).getPriorityLevel();
-
- // first find available-consumer on higher level unless currentIndex is not on highest level which is 0
- if (currentRoundRobinConsumerPriority != 0) {
- int higherPriorityConsumerIndex = getConsumerFromHigherPriority(currentRoundRobinConsumerPriority);
- if (higherPriorityConsumerIndex != -1) {
- currentConsumerRoundRobinIndex = higherPriorityConsumerIndex + 1;
- return consumerList.get(higherPriorityConsumerIndex);
- }
- }
-
- // currentIndex is already on highest level or couldn't find consumer on higher level so, find consumer on same
- // or lower level
- int availableConsumerIndex = getNextConsumerFromSameOrLowerLevel(currentConsumerRoundRobinIndex);
+ int availableConsumerIndex = consumerPrioritySelector.select(currentConsumerRoundRobinIndex);
if (availableConsumerIndex != -1) {
currentConsumerRoundRobinIndex = availableConsumerIndex + 1;
return consumerList.get(availableConsumerIndex);
}
-
- // couldn't find available consumer
return null;
}
@@ -173,84 +175,4 @@ public abstract class AbstractDispatcherMultipleConsumers extends AbstractBaseDi
}
- /**
- * Finds index of first available consumer which has higher priority then given targetPriority.
- *
- * @param targetPriority
- * @return -1 if couldn't find any available consumer
- */
- private int getConsumerFromHigherPriority(int targetPriority) {
- for (int i = 0; i < currentConsumerRoundRobinIndex; i++) {
- Consumer consumer = consumerList.get(i);
- if (consumer.getPriorityLevel() < targetPriority) {
- if (isConsumerAvailable(consumerList.get(i))) {
- return i;
- }
- } else {
- break;
- }
- }
- return -1;
- }
-
- /**
- * Finds index of round-robin available consumer that present on same level as consumer on
- * currentRoundRobinIndex if doesn't find consumer on same level then it finds first available consumer on lower
- * priority level else returns
- * index=-1 if couldn't find any available consumer in the list.
- *
- * @param currentRoundRobinIndex
- * @return
- */
- private int getNextConsumerFromSameOrLowerLevel(int currentRoundRobinIndex) {
- Consumer currentRRConsumer = consumerList.get(currentRoundRobinIndex);
- if (isConsumerAvailable(currentRRConsumer)) {
- return currentRoundRobinIndex;
- }
-
- // scan the consumerList, if consumer in currentRoundRobinIndex is unavailable
- int targetPriority = currentRRConsumer.getPriorityLevel();
- int scanIndex = currentRoundRobinIndex + 1;
- int endPriorityLevelIndex = currentRoundRobinIndex;
- do {
- Consumer scanConsumer = scanIndex < consumerList.size() ? consumerList.get(scanIndex)
- : null /* reached to last consumer of list */;
-
- // if reached to last consumer of list then check from beginning to currentRRIndex of the list
- if (scanConsumer == null || scanConsumer.getPriorityLevel() != targetPriority) {
- endPriorityLevelIndex = scanIndex; // last consumer on this level
- scanIndex = getFirstConsumerIndexOfPriority(targetPriority);
- } else {
- if (isConsumerAvailable(scanConsumer)) {
- return scanIndex;
- }
- scanIndex++;
- }
- } while (scanIndex != currentRoundRobinIndex);
-
- // it means: didn't find consumer in the same priority-level so, check available consumer lower than this level
- for (int i = endPriorityLevelIndex; i < consumerList.size(); i++) {
- if (isConsumerAvailable(consumerList.get(i))) {
- return i;
- }
- }
-
- return -1;
- }
-
- /**
- * Finds index of first consumer in list which has same priority as given targetPriority.
- *
- * @param targetPriority
- * @return
- */
- private int getFirstConsumerIndexOfPriority(int targetPriority) {
- for (int i = 0; i < consumerList.size(); i++) {
- if (consumerList.get(i).getPriorityLevel() == targetPriority) {
- return i;
- }
- }
- return -1;
- }
-
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ConsumerPrioritySelector.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ConsumerPrioritySelector.java
new file mode 100644
index 000000000000..a69ddd5fd11f
--- /dev/null
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ConsumerPrioritySelector.java
@@ -0,0 +1,168 @@
+/*
+ * 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 com.google.common.annotations.VisibleForTesting;
+import it.unimi.dsi.fastutil.ints.Int2ObjectMap;
+import it.unimi.dsi.fastutil.ints.Int2ObjectOpenHashMap;
+import java.util.List;
+import java.util.function.Predicate;
+import java.util.function.ToIntFunction;
+import javax.annotation.concurrent.NotThreadSafe;
+import org.apache.commons.lang3.mutable.MutableInt;
+
+/**
+ * Selects from consumers ordered by increasing priority number, with round-robin selection within a priority.
+ * The caller owns the cursor and must serialize selection and membership changes using the same lock.
+ * After construction, membership changes must go through this selector; sorting the list is allowed.
+ * Consumer priorities must remain immutable while registered.
+ *
+ * <p>The type parameter lets unit tests and JMH benchmarks use lightweight consumer state to isolate
+ * selection from broker setup and other {@link Consumer} behavior. Production uses {@code Consumer}
+ * with {@link Consumer#getPriorityLevel()} as the priority accessor.
+ *
+ * @param <T> consumer representation: {@code Consumer} in production, lightweight fixtures in tests and benchmarks
+ */
+@NotThreadSafe
+final class ConsumerPrioritySelector<T> {
+ private final List<T> consumerList;
+ private final ToIntFunction<T> priority;
+ private final Predicate<T> available;
+ private final Int2ObjectMap<MutableInt> priorityCounts = new Int2ObjectOpenHashMap<>();
+
+ ConsumerPrioritySelector(List<T> consumerList, ToIntFunction<T> priority, Predicate<T> available) {
+ this.consumerList = consumerList;
+ this.priority = priority;
+ this.available = available;
+ consumerList.forEach(this::incrementPriorityCount);
+ }
+
+ void add(T consumer) {
+ consumerList.add(consumer);
+ incrementPriorityCount(consumer);
+ }
+
+ void remove(T consumer) {
+ int index = consumerList.indexOf(consumer);
+ if (index >= 0) {
+ // Equality can match a replacement with a different priority. Count the actual removed instance.
+ decrementPriorityCount(consumerList.remove(index));
+ }
+ }
+
+ void removeIf(Predicate<T> predicate) {
+ consumerList.removeIf(predicate);
+ // Bulk removal repairs inconsistent dispatcher membership. Rebuild from survivors so even an
+ // unregistered duplicate in the list cannot leave stale counts or prevent topic deletion.
+ priorityCounts.clear();
+ consumerList.forEach(this::incrementPriorityCount);
+ }
+
+ private void incrementPriorityCount(T consumer) {
+ priorityCounts.compute(priority.applyAsInt(consumer), (level, count) -> {
+ if (count == null) {
+ return new MutableInt(1);
+ }
+ count.increment();
+ return count;
+ });
+ }
+
+ private void decrementPriorityCount(T consumer) {
+ priorityCounts.compute(priority.applyAsInt(consumer),
+ (level, count) -> count.decrementAndGet() == 0 ? null : count);
+ }
+
+ @VisibleForTesting
+ int priorityLevelCount() {
+ return priorityCounts.size();
+ }
+
+ // The caller supplies an in-range cursor and advances it only after a successful selection.
+ int select(int cursor) {
+ if (priorityCounts.size() == 1) {
+ return selectSamePriority(cursor);
+ }
+ T current = consumerList.get(cursor);
+ int targetPriority = priority.applyAsInt(current);
+ int firstOnLevel = 0;
+
+ if (targetPriority != 0) {
+ // Remember where this level starts so wrapping does not rescan all higher priorities.
+ for (; firstOnLevel < cursor; firstOnLevel++) {
+ T consumer = consumerList.get(firstOnLevel);
+ if (priority.applyAsInt(consumer) >= targetPriority) {
+ break;
+ }
+ if (available.test(consumer)) {
+ return firstOnLevel;
+ }
+ }
+ }
+
+ if (available.test(current)) {
+ return cursor;
+ }
+
+ int size = consumerList.size();
+ int endOfLevel = cursor + 1;
+ for (; endOfLevel < size; endOfLevel++) {
+ T consumer = consumerList.get(endOfLevel);
+ if (priority.applyAsInt(consumer) != targetPriority) {
+ break;
+ }
+ if (available.test(consumer)) {
+ return endOfLevel;
+ }
+ }
+
+ if (targetPriority == 0) {
+ // Normally zero is the first level. Preserve the existing treatment of negative priorities too.
+ while (priority.applyAsInt(consumerList.get(firstOnLevel)) != targetPriority) {
+ firstOnLevel++;
+ }
+ }
+ for (int i = firstOnLevel; i < cursor; i++) {
+ if (available.test(consumerList.get(i))) {
+ return i;
+ }
+ }
+ for (int i = endOfLevel; i < size; i++) {
+ if (available.test(consumerList.get(i))) {
+ return i;
+ }
+ }
+ return -1;
+ }
+
+ private int selectSamePriority(int cursor) {
+ int size = consumerList.size();
+ for (int i = cursor; i < size; i++) {
+ if (available.test(consumerList.get(i))) {
+ return i;
+ }
+ }
+ for (int i = 0; i < cursor; i++) {
+ if (available.test(consumerList.get(i))) {
+ return i;
+ }
+ }
+ return -1;
+ }
+}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentDispatcherMultipleConsumers.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentDispatcherMultipleConsumers.java
index d70cf5baef14..2563a3957415 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentDispatcherMultipleConsumers.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentDispatcherMultipleConsumers.java
@@ -92,7 +92,7 @@ public class NonPersistentDispatcherMultipleConsumers extends AbstractDispatcher
return FutureUtil.failedFuture(new ConsumerBusyException("Subscription reached max consumers limit"));
}
- consumerList.add(consumer);
+ addConsumerToList(consumer);
consumerSet.add(consumer);
return CompletableFuture.completedFuture(null);
}
@@ -105,7 +105,7 @@ public class NonPersistentDispatcherMultipleConsumers extends AbstractDispatcher
@Override
public synchronized void removeConsumer(Consumer consumer) throws BrokerServiceException {
if (consumerSet.removeAll(consumer) == 1) {
- consumerList.remove(consumer);
+ removeConsumerFromList(consumer);
log.info().attr("consumer", consumer).log("Removed consumer");
if (consumerList.isEmpty()) {
if (closeFuture != null) {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentStickyKeyDispatcherMultipleConsumers.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentStickyKeyDispatcherMultipleConsumers.java
index 657a6fcf4d17..047a836b2d3f 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentStickyKeyDispatcherMultipleConsumers.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentStickyKeyDispatcherMultipleConsumers.java
@@ -101,7 +101,7 @@ public class NonPersistentStickyKeyDispatcherMultipleConsumers extends NonPersis
if (ex != null) {
synchronized (NonPersistentStickyKeyDispatcherMultipleConsumers.this) {
consumerSet.removeAll(consumer);
- consumerList.remove(consumer);
+ removeConsumerFromList(consumer);
}
throw FutureUtil.wrapToCompletionException(ex);
}
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 359ab62e6aa8..760f764c573a 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
@@ -224,7 +224,7 @@ public class PersistentDispatcherMultipleConsumers extends AbstractPersistentDis
.log("Attempting to add a consumer that already registered");
}
- consumerList.add(consumer);
+ addConsumerToList(consumer);
if (consumerList.size() > 1
&& consumer.getPriorityLevel() < consumerList.get(consumerList.size() - 2).getPriorityLevel()) {
consumerList.sort(Comparator.comparingInt(Consumer::getPriorityLevel));
@@ -246,7 +246,7 @@ public class PersistentDispatcherMultipleConsumers extends AbstractPersistentDis
// 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);
+ removeConsumerFromList(consumer);
log.info()
.attr("consumer", consumer)
.attr("pendingAcks", consumer.getPendingAcks().size())
@@ -284,7 +284,7 @@ public class PersistentDispatcherMultipleConsumers extends AbstractPersistentDis
// 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));
+ removeConsumersFromList(c -> consumer.equals(c));
if (consumerList.isEmpty()) {
clearComponentsAfterRemovedAllConsumers();
}
@@ -293,7 +293,7 @@ public class PersistentDispatcherMultipleConsumers extends AbstractPersistentDis
protected synchronized void internalRemoveConsumer(Consumer consumer) {
consumerSet.removeAll(consumer);
- consumerList.remove(consumer);
+ removeConsumerFromList(consumer);
}
protected synchronized void 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 f59d574ed551..1f5e70a741bc 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
@@ -214,7 +214,7 @@ public class PersistentDispatcherMultipleConsumersClassic extends AbstractPersis
.log("Attempting to add a consumer that already registered");
}
- consumerList.add(consumer);
+ addConsumerToList(consumer);
if (consumerList.size() > 1
&& consumer.getPriorityLevel() < consumerList.get(consumerList.size() - 2).getPriorityLevel()) {
consumerList.sort(Comparator.comparingInt(Consumer::getPriorityLevel));
@@ -236,7 +236,7 @@ public class PersistentDispatcherMultipleConsumersClassic extends AbstractPersis
// 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);
+ removeConsumerFromList(consumer);
log.info()
.attr("consumer", consumer)
.attr("size", consumer.getPendingAcks().size())
@@ -267,7 +267,7 @@ public class PersistentDispatcherMultipleConsumersClassic extends AbstractPersis
*/
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));
+ removeConsumersFromList(c -> consumer.equals(c));
if (consumerList.isEmpty()) {
clearComponentsAfterRemovedAllConsumers();
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersClassic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersClassic.java
index 624c03964180..223561177864 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersClassic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersClassic.java
@@ -160,7 +160,7 @@ public class PersistentStickyKeyDispatcherMultipleConsumersClassic
if (ex != null) {
synchronized (PersistentStickyKeyDispatcherMultipleConsumersClassic.this) {
consumerSet.removeAll(consumer);
- consumerList.remove(consumer);
+ removeConsumerFromList(consumer);
}
throw FutureUtil.wrapToCompletionException(ex);
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ConsumerPrioritySelectorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ConsumerPrioritySelectorTest.java
new file mode 100644
index 000000000000..bbe3f91c56fe
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ConsumerPrioritySelectorTest.java
@@ -0,0 +1,233 @@
+/*
+ * 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 static org.assertj.core.api.Assertions.assertThatThrownBy;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import org.testng.annotations.Test;
+
+public class ConsumerPrioritySelectorTest {
+ @Test
+ public void selectionMatchesPriorityAndRoundRobinOrder() {
+ // Every grouping of eight consumers, every availability mask, and every cursor.
+ for (int boundaries = 0; boundaries < 128; boundaries++) {
+ List<Slot> consumers = new CopyOnWriteArrayList<>();
+ int priority = (boundaries & 1) == 0 ? 0 : 3;
+ for (int i = 0; i < 8; i++) {
+ if (i > 0 && (boundaries & (1 << (i - 1))) != 0) {
+ priority += 3;
+ }
+ consumers.add(new Slot(i, priority));
+ }
+ List<Integer> visited = new ArrayList<>();
+ ConsumerPrioritySelector<Slot> selector = new ConsumerPrioritySelector<>(consumers, c -> c.priority, c -> {
+ visited.add(c.index);
+ return c.available;
+ });
+ for (int cursor = 0; cursor < consumers.size(); cursor++) {
+ int start = cursor;
+ int currentPriority = consumers.get(cursor).priority;
+ List<Slot> ordered = new ArrayList<>(consumers);
+ ordered.sort(Comparator.comparingInt((Slot c) -> c.priority)
+ .thenComparingInt(c -> c.priority == currentPriority && c.index < start ? 1 : 0)
+ .thenComparingInt(c -> c.index));
+ for (int mask = 0; mask < 256; mask++) {
+ for (Slot c : consumers) {
+ c.available = (mask & (1 << c.index)) != 0;
+ }
+ List<Integer> expectedVisits = new ArrayList<>();
+ int expected = -1;
+ for (Slot c : ordered) {
+ expectedVisits.add(c.index);
+ if (c.available) {
+ expected = c.index;
+ break;
+ }
+ }
+ visited.clear();
+ assertThat(selector.select(cursor)).isEqualTo(expected);
+ assertThat(visited).isEqualTo(expectedVisits);
+ }
+ }
+ }
+ }
+
+ @Test
+ public void observesMembershipAndPermitChanges() {
+ List<Slot> consumers = new CopyOnWriteArrayList<>();
+ Slot first = new Slot(0, 0);
+ Slot second = new Slot(1, 3);
+ consumers.add(first);
+ consumers.add(second);
+ ConsumerPrioritySelector<Slot> selector = new ConsumerPrioritySelector<>(consumers, c -> c.priority,
+ c -> c.available);
+ second.available = true;
+ assertThat(selector.select(0)).isEqualTo(1);
+ first.available = true;
+ assertThat(selector.select(1)).isZero();
+ selector.remove(first);
+ assertThat(selector.select(0)).isZero();
+ second.available = false;
+ assertThat(selector.select(0)).isEqualTo(-1);
+ selector.add(first);
+ consumers.sort(Comparator.comparingInt(c -> c.priority));
+ assertThat(selector.select(1)).isZero();
+ }
+
+ @Test
+ public void zeroPriorityRetainsItsSpecialTreatment() {
+ Slot negative = new Slot(0, -1);
+ Slot zero = new Slot(1, 0);
+ Slot lower = new Slot(2, 1);
+ List<Slot> consumers = new CopyOnWriteArrayList<>(List.of(negative, zero, lower));
+ ConsumerPrioritySelector<Slot> selector = new ConsumerPrioritySelector<>(consumers, c -> c.priority,
+ c -> c.available);
+ negative.available = true;
+ lower.available = true;
+ // The original selector skips its higher-priority scan when the cursor priority is zero.
+ assertThat(selector.select(1)).isEqualTo(2);
+ lower.available = false;
+ assertThat(selector.select(1)).isEqualTo(-1);
+ zero.available = true;
+ assertThat(selector.select(1)).isEqualTo(1);
+ assertThat(selector.select(2)).isZero();
+ }
+
+ @Test
+ public void fiftyConsumersAtTheSamePriority() {
+ for (int priority : new int[]{0, 3}) {
+ List<Slot> consumers = new CopyOnWriteArrayList<>();
+ for (int i = 0; i < 50; i++) {
+ Slot consumer = new Slot(i, priority);
+ consumer.available = true;
+ consumers.add(consumer);
+ }
+ ConsumerPrioritySelector<Slot> selector = new ConsumerPrioritySelector<>(consumers, c -> c.priority,
+ c -> c.available);
+ int cursor = 25;
+ for (int i = 0; i < 150; i++) {
+ int expected = (25 + i) % 50;
+ int selected = selector.select(cursor % consumers.size());
+ assertThat(selected).isEqualTo(expected);
+ cursor = selected + 1;
+ }
+ // Drain the group starting in its middle, including wrapping to earlier consumers.
+ for (int i = 0; i < 50; i++) {
+ int selected = selector.select(cursor % consumers.size());
+ assertThat(selected).isEqualTo((25 + i) % 50);
+ consumers.get(selected).available = false;
+ cursor = selected + 1;
+ }
+ assertThat(selector.select(cursor)).isEqualTo(-1);
+ consumers.get(7).available = true;
+ assertThat(selector.select(cursor)).isEqualTo(7);
+ }
+ }
+
+ @Test
+ public void priorityCountsFollowActualRemovals() {
+ List<Slot> consumers = new CopyOnWriteArrayList<>();
+ ConsumerPrioritySelector<Slot> selector = new ConsumerPrioritySelector<>(consumers, c -> c.priority,
+ c -> c.available);
+ Slot first = new Slot(0, 0);
+ first.available = true;
+ Slot second = new Slot(1, 3);
+ Slot third = new Slot(2, 3);
+ assertThat(selector.priorityLevelCount()).isZero();
+ selector.add(first);
+ selector.add(second);
+ selector.add(third);
+ assertThat(selector.priorityLevelCount()).isEqualTo(2);
+ // Equal protocol identity does not imply equal priority: decrement the registered instance's count.
+ selector.remove(new Slot(1, 99));
+ assertThat(selector.priorityLevelCount()).isEqualTo(2);
+ selector.removeIf(c -> c.priority == 3);
+ assertThat(selector.priorityLevelCount()).isEqualTo(1);
+ assertThat(selector.select(0)).isZero();
+ selector.remove(first);
+ assertThat(selector.priorityLevelCount()).isZero();
+ selector.remove(first);
+ assertThat(selector.priorityLevelCount()).isZero();
+ selector.add(second);
+ selector.add(second);
+ selector.remove(second);
+ assertThat(selector.priorityLevelCount()).isEqualTo(1);
+ selector.removeIf(c -> true);
+ assertThat(selector.priorityLevelCount()).isZero();
+ }
+
+ @Test
+ public void bulkRemovalRepairsInconsistentMembership() {
+ List<Slot> consumers = new CopyOnWriteArrayList<>();
+ ConsumerPrioritySelector<Slot> selector = new ConsumerPrioritySelector<>(consumers, c -> c.priority,
+ c -> c.available);
+ Slot consumer = new Slot(0, 0);
+ selector.add(consumer);
+ // Model the existing dispatcher recovery path for a list/set mismatch.
+ consumers.add(consumer);
+ selector.remove(consumer);
+ selector.removeIf(c -> c.equals(consumer));
+ assertThat(consumers).isEmpty();
+ assertThat(selector.priorityLevelCount()).isZero();
+ }
+
+ @Test
+ public void failedBulkRemovalPreservesCounts() {
+ List<Slot> consumers = new CopyOnWriteArrayList<>();
+ ConsumerPrioritySelector<Slot> selector = new ConsumerPrioritySelector<>(consumers, c -> c.priority,
+ c -> c.available);
+ selector.add(new Slot(0, 0));
+ selector.add(new Slot(1, 3));
+ assertThatThrownBy(() -> selector.removeIf(c -> {
+ if (c.index == 1) {
+ throw new IllegalStateException("predicate failed");
+ }
+ return true;
+ })).isInstanceOf(IllegalStateException.class);
+ assertThat(consumers).hasSize(2);
+ assertThat(selector.priorityLevelCount()).isEqualTo(2);
+ selector.removeIf(c -> true);
+ assertThat(selector.priorityLevelCount()).isZero();
+ }
+
+ private static final class Slot {
+ private final int index;
+ private final int priority;
+ private boolean available;
+
+ private Slot(int index, int priority) {
+ this.index = index;
+ this.priority = priority;
+ }
+
+ @Override
+ public boolean equals(Object other) {
+ return other instanceof Slot slot && index == slot.index;
+ }
+
+ @Override
+ public int hashCode() {
+ return index;
+ }
+ }
+}