Skip to content

[fix][broker] Prevent stale consumer removal from removing replacements #26652

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
Denovo1998 wants to merge 2 commits into
base: master
Choose a base branch
from
Open

[fix][broker] Prevent stale consumer removal from removing replacements #26652

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
dc7dcf3
[fix][broker] Prevent stale consumer removal from removing replacements
Denovo1998 Sep 19, 2026
cc0311f
Ensure sticky dispatcher cleanup handles unregistered consumers and s…
Denovo1998 Sep 20, 2026
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
15 changes: 15 additions & 0 deletions ...r/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherMultipleConsumers.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,10 @@ protected final synchronized void removeConsumerFromList(Consumer consumer) {
consumerPrioritySelector.remove(consumer);
}

protected final synchronized void removeConsumerInstanceFromList(Consumer consumer) {
consumerPrioritySelector.removeInstance(consumer);
}

protected final synchronized void removeConsumersFromList(Predicate<Consumer> predicate) {
consumerPrioritySelector.removeIf(predicate);
}
Expand Down Expand Up @@ -88,6 +92,17 @@ protected final boolean containsConsumerInstance(Consumer consumer) {
return consumerSetImpl.indexExists(index) && consumerSetImpl.indexGet(index) == consumer;
}

/**
* Removes only the exact registered instance. The caller must hold the dispatcher monitor.
*/
protected final boolean removeConsumerInstance(Consumer consumer) {
if (!containsConsumerInstance(consumer)) {
return false;
}
consumerSet.removeAll(consumer);
return true;
}

public boolean isClosed() {
return isClosed == TRUE;
}
Expand Down
9 changes: 9 additions & 0 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ConsumerPrioritySelector.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -66,6 +66,15 @@ void remove(T consumer) {
}
}

void removeInstance(T consumer) {
for (int index = 0; index < consumerList.size(); index++) {
if (consumerList.get(index) == consumer) {
decrementPriorityCount(consumerList.remove(index));
return;
}
}
}

void removeIf(Predicate<T> predicate) {
consumerList.removeIf(predicate);
// Bulk removal repairs inconsistent dispatcher membership. Rebuild from survivors so even an
Expand Down
2 changes: 1 addition & 1 deletion ...in/java/org/apache/pulsar/broker/service/HashRangeExclusiveStickyKeyConsumerSelector.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,7 +81,7 @@ private synchronized Optional<ImpactedConsumersResult> internalAddConsumer(Consu

@Override
public synchronized Optional<ImpactedConsumersResult> removeConsumer(Consumer consumer) {
rangeMap.entrySet().removeIf(entry -> entry.getValue().getRight().equals(consumer));
rangeMap.entrySet().removeIf(entry -> entry.getValue().getRight() == consumer);
return Optional.empty();
}

Expand Down
10 changes: 5 additions & 5 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 @@ -241,12 +241,12 @@ protected boolean isConsumersExceededOnSubscription() {

@Override
public synchronized void removeConsumer(Consumer consumer) throws BrokerServiceException {
if (consumerSet.removeAll(consumer) == 1) {
if (removeConsumerInstance(consumer)) {
// 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());
removeConsumerFromList(consumer);
removeConsumerInstanceFromList(consumer);
log.info()
.attr("consumer", consumer)
.attr("pendingAcks", consumer.getPendingAcks().size())
Expand Down Expand Up @@ -284,16 +284,16 @@ public synchronized void removeConsumer(Consumer consumer) throws BrokerServiceE
// 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.
removeConsumersFromList(c -> consumer.equals(c));
removeConsumersFromList(c -> c == consumer);
if (consumerList.isEmpty()) {
clearComponentsAfterRemovedAllConsumers();
}
}
}

protected synchronized void internalRemoveConsumer(Consumer consumer) {
consumerSet.removeAll(consumer);
removeConsumerFromList(consumer);
removeConsumerInstance(consumer);
removeConsumerInstanceFromList(consumer);
}

protected synchronized void clearComponentsAfterRemovedAllConsumers() {
Expand Down
6 changes: 3 additions & 3 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 @@ -231,12 +231,12 @@ protected boolean isConsumersExceededOnSubscription() {

@Override
public synchronized void removeConsumer(Consumer consumer) throws BrokerServiceException {
if (consumerSet.removeAll(consumer) == 1) {
if (removeConsumerInstance(consumer)) {
// 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());
removeConsumerFromList(consumer);
removeConsumerInstanceFromList(consumer);
log.info()
.attr("consumer", consumer)
.attr("size", consumer.getPendingAcks().size())
Expand Down Expand Up @@ -267,7 +267,7 @@ public synchronized void removeConsumer(Consumer consumer) throws BrokerServiceE
*/
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.
removeConsumersFromList(c -> consumer.equals(c));
removeConsumersFromList(c -> c == consumer);
if (consumerList.isEmpty()) {
clearComponentsAfterRemovedAllConsumers();
}
Expand Down
7 changes: 7 additions & 0 deletions ...ache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -226,6 +226,13 @@ private synchronized void registerDrainingHashes(Consumer skipConsumer,

@Override
public synchronized void removeConsumer(Consumer consumer) throws BrokerServiceException {
// A STICKY addition can finish its liveness check after dispatcher removal. Its selector removes by
// identity, so repeat its cleanup even for an unregistered consumer without affecting replacements.
if (keySharedMode != KeySharedMode.STICKY && !containsConsumerInstance(consumer)) {
// Let the superclass repair stale list membership without touching a replacement's selector state.
super.removeConsumer(consumer);
return;

@lhotari lhotari Sep 19, 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.

[BUG] Keep cleanup for STICKY additions that finish after removal

An unregistered consumer can still have selector state: HashRangeExclusiveStickyKeyConsumerSelector.addConsumer waits on a conflicting consumer's liveness check before installing its range (HashRangeExclusiveStickyKeyConsumerSelector.java:59-79).

Consider a pending STICKY subscribe on another connection that times out while this check is outstanding. A concurrent cursor reset removes both consumers from the dispatcher. If the liveness check then completes, the selector can install the removed consumer's range. The original subscribe future is already exceptional, so ServerCnx calls consumer.close() again (ServerCnx.java:2152-2155). This new early return skips the selector cleanup that previously ran on that second close, leaving an orphan range that blocks later same-range subscriptions. The classic guard at PersistentStickyKeyDispatcherMultipleConsumersClassic.java:237-240 has the same issue.

Please invalidate pending additions when removing a consumer, or preserve exact-instance selector cleanup for this case while protecting equal replacements. A regression test can hold the STICKY liveness future, remove both consumers, complete the add, then repeat the close and verify that a new same-range subscribe succeeds.

@Denovo1998 Denovo1998 Sep 20, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

@lhotari Indeed, there is this issue. I just dealt with it, could you please review it again and take a look?

}
// The consumer must be removed from the selector before calling the superclass removeConsumer method.
Optional<ImpactedConsumersResult> impactedConsumers = selector.removeConsumer(consumer);
super.removeConsumer(consumer);
Expand Down
13 changes: 10 additions & 3 deletions ...lsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersClassic.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -159,8 +159,8 @@ public synchronized CompletableFuture<Void> addConsumer(Consumer consumer) {
selector.addConsumer(consumer).handle((result, ex) -> {
if (ex != null) {
synchronized (PersistentStickyKeyDispatcherMultipleConsumersClassic.this) {
consumerSet.removeAll(consumer);
removeConsumerFromList(consumer);
removeConsumerInstance(consumer);
removeConsumerInstanceFromList(consumer);
}
throw FutureUtil.wrapToCompletionException(ex);
}
Expand Down Expand Up @@ -234,6 +234,13 @@ private void sortRecentlyJoinedConsumersIfNeeded() {

@Override
public synchronized void removeConsumer(Consumer consumer) throws BrokerServiceException {
// A STICKY addition can finish its liveness check after dispatcher removal. Its selector removes by
// identity, so repeat its cleanup even for an unregistered consumer without affecting replacements.
if (keySharedMode != KeySharedMode.STICKY && !containsConsumerInstance(consumer)) {
// Let the superclass repair stale list membership without touching a replacement's selector state.
super.removeConsumer(consumer);
return;
}
// The consumer must be removed from the selector before calling the superclass removeConsumer method.
// In the superclass removeConsumer method, the pending acks that the consumer has are added to
// redeliveryMessages. If the consumer has not been removed from the selector at this point,
Expand All @@ -243,7 +250,7 @@ public synchronized void removeConsumer(Consumer consumer) throws BrokerServiceE
selector.removeConsumer(consumer);
super.removeConsumer(consumer);
if (recentlyJoinedConsumers != null) {
recentlyJoinedConsumers.remove(consumer);
recentlyJoinedConsumers.keySet().removeIf(c -> c == consumer);
if (consumerList.size() == 1) {
recentlyJoinedConsumers.clear();
}
Expand Down
16 changes: 16 additions & 0 deletions ...r-broker/src/test/java/org/apache/pulsar/broker/service/ConsumerPrioritySelectorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -176,6 +176,22 @@ public void priorityCountsFollowActualRemovals() {
assertThat(selector.priorityLevelCount()).isZero();
}

@Test
public void instanceRemovalPreservesEqualReplacement() {
List<Slot> consumers = new CopyOnWriteArrayList<>();
ConsumerPrioritySelector<Slot> selector = new ConsumerPrioritySelector<>(consumers, c -> c.priority,
c -> c.available);
Slot original = new Slot(0, 0);
Slot replacement = new Slot(0, 3);
selector.add(replacement);
selector.removeInstance(original);
assertThat(consumers).singleElement().isSameAs(replacement);
assertThat(selector.priorityLevelCount()).isEqualTo(1);
selector.removeInstance(replacement);
assertThat(consumers).isEmpty();
assertThat(selector.priorityLevelCount()).isZero();
}

@Test
public void bulkRemovalRepairsInconsistentMembership() {
List<Slot> consumers = new CopyOnWriteArrayList<>();
Expand Down
Loading