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/10/16 22:31:28 UTC
(pulsar) 05/24: [fix][broker] Prevent stale consumer removal from removing replacements (#26652)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0-backports-2026-10-17
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit ebdeb3f2dd0910be78f29b4dc09a5ce1fbcfdc2c
Author: sinan liu <li...@gmail.com>
AuthorDate: Thu Oct 15 03:42:19 2026 +0800
[fix][broker] Prevent stale consumer removal from removing replacements (#26652)
(cherry picked from commit 668c16d15e0a3ab0af6fd7e874519968a7c88c53)
---
.../AbstractDispatcherMultipleConsumers.java | 18 +++
...ashRangeExclusiveStickyKeyConsumerSelector.java | 2 +-
.../PersistentDispatcherMultipleConsumers.java | 10 +-
...rsistentDispatcherMultipleConsumersClassic.java | 6 +-
...istentStickyKeyDispatcherMultipleConsumers.java | 7 +
...dSubscriptionUnackedMessagesAccountingTest.java | 143 +++++++++++++++++++++
6 files changed, 177 insertions(+), 9 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 5d1dd116e85b..9a415ace99a0 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
@@ -47,6 +47,13 @@ public abstract class AbstractDispatcherMultipleConsumers extends AbstractBaseDi
super(subscription, serviceConfig);
}
+ /**
+ * Removes only the exact consumer instance from {@link #consumerList}, which would otherwise match a replacement
+ * consumer that reuses the same protocol identity via {@link Consumer#equals(Object)}.
+ */
+ protected final synchronized void removeConsumerInstanceFromList(Consumer consumer) {
+ consumerList.removeIf(c -> c == consumer);
+ }
public boolean isConsumerConnected() {
return !consumerList.isEmpty();
}
@@ -71,6 +78,17 @@ public abstract class AbstractDispatcherMultipleConsumers extends AbstractBaseDi
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;
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/HashRangeExclusiveStickyKeyConsumerSelector.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/HashRangeExclusiveStickyKeyConsumerSelector.java
index 7fb998381972..7ed3db4007d3 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/HashRangeExclusiveStickyKeyConsumerSelector.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/HashRangeExclusiveStickyKeyConsumerSelector.java
@@ -81,7 +81,7 @@ public class HashRangeExclusiveStickyKeyConsumerSelector implements StickyKeyCon
@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();
}
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 fdfc7d116af3..32f9828aa835 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
@@ -229,12 +229,12 @@ public class PersistentDispatcherMultipleConsumers extends AbstractPersistentDis
@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());
- consumerList.remove(consumer);
+ removeConsumerInstanceFromList(consumer);
log.info("Removed consumer {} with pending {} acks", consumer, consumer.getPendingAcks().size());
if (consumerList.isEmpty()) {
clearComponentsAfterRemovedAllConsumers();
@@ -272,7 +272,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));
+ consumerList.removeIf(c -> c == consumer);
if (consumerList.isEmpty()) {
clearComponentsAfterRemovedAllConsumers();
}
@@ -280,8 +280,8 @@ public class PersistentDispatcherMultipleConsumers extends AbstractPersistentDis
}
protected synchronized void internalRemoveConsumer(Consumer consumer) {
- consumerSet.removeAll(consumer);
- consumerList.remove(consumer);
+ removeConsumerInstance(consumer);
+ removeConsumerInstanceFromList(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 82cc5aca3bfa..4d0f9f7159fa 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
@@ -218,12 +218,12 @@ public class PersistentDispatcherMultipleConsumersClassic extends AbstractPersis
@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());
- consumerList.remove(consumer);
+ removeConsumerInstanceFromList(consumer);
log.info("Removed consumer {} with pending {} acks", consumer, consumer.getPendingAcks().size());
if (consumerList.isEmpty()) {
clearComponentsAfterRemovedAllConsumers();
@@ -253,7 +253,7 @@ public class PersistentDispatcherMultipleConsumersClassic extends AbstractPersis
*/
log.error("[{}] Trying to remove a non-connected consumer: {}", name, consumer);
// The debit belongs to the removal that unregisters the consumer; do not repeat it here.
- consumerList.removeIf(c -> consumer.equals(c));
+ consumerList.removeIf(c -> c == consumer);
if (consumerList.isEmpty()) {
clearComponentsAfterRemovedAllConsumers();
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java
index 6d36a7b0dd31..c449c21df20c 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java
@@ -314,6 +314,13 @@ public class PersistentStickyKeyDispatcherMultipleConsumers extends PersistentDi
@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.
Optional<ImpactedConsumersResult> impactedConsumers = selector.removeConsumer(consumer);
super.removeConsumer(consumer);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SharedSubscriptionUnackedMessagesAccountingTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SharedSubscriptionUnackedMessagesAccountingTest.java
index ad9135b0df19..0599a67a124b 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SharedSubscriptionUnackedMessagesAccountingTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SharedSubscriptionUnackedMessagesAccountingTest.java
@@ -19,17 +19,29 @@
package org.apache.pulsar.broker.service;
import static org.assertj.core.api.Assertions.assertThat;
+import io.netty.channel.Channel;
+import java.util.Collections;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import org.apache.pulsar.broker.service.persistent.AbstractPersistentDispatcherMultipleConsumers;
import org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers;
import org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumersClassic;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+import org.apache.pulsar.client.api.ConsumerBuilder;
+import org.apache.pulsar.client.api.KeySharedPolicy;
+import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.ProducerConsumerBase;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.Range;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.common.api.proto.CommandSubscribe.InitialPosition;
+import org.awaitility.Awaitility;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.DataProvider;
import org.testng.annotations.Factory;
import org.testng.annotations.Test;
@@ -121,6 +133,137 @@ public class SharedSubscriptionUnackedMessagesAccountingTest extends ProducerCon
}
}
+ @DataProvider
+ public Object[][] subscriptionTypes() {
+ return new Object[][] {
+ {SubscriptionType.Shared, null},
+ {SubscriptionType.Key_Shared, KeySharedPolicy.autoSplitHashRange()},
+ {SubscriptionType.Key_Shared, KeySharedPolicy.stickyHashRange().ranges(Range.of(0, 65535))}};
+ }
+
+ @Test(dataProvider = "subscriptionTypes", timeOut = 60_000)
+ public void testStaleRemovalDoesNotRemoveEqualReplacement(SubscriptionType subscriptionType,
+ KeySharedPolicy keySharedPolicy) throws Exception {
+ String topicName = newTopicName();
+ ConsumerBuilder<String> consumerBuilder = pulsarClient.newConsumer(Schema.STRING)
+ .topic(topicName).subscriptionName(SUBSCRIPTION).subscriptionType(subscriptionType)
+ .consumerName("replacement-test");
+ if (keySharedPolicy != null) {
+ consumerBuilder.keySharedPolicy(keySharedPolicy);
+ }
+ try (Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+ .topic(topicName).enableBatching(false).create();
+ org.apache.pulsar.client.api.Consumer<String> client = consumerBuilder.subscribe()) {
+ producer.send("unacked");
+ assertThat(client.receive(5, TimeUnit.SECONDS)).isNotNull();
+ BrokerService brokerService = pulsar.getBrokerService();
+ PersistentTopic topic = (PersistentTopic) brokerService.getTopicReference(topicName).orElseThrow();
+ AbstractPersistentDispatcherMultipleConsumers dispatcher =
+ (AbstractPersistentDispatcherMultipleConsumers) topic.getSubscription(SUBSCRIPTION).getDispatcher();
+ Consumer original = dispatcher.getConsumers().get(0);
+ synchronized (dispatcher) {
+ assertThat(original.getUnackedMessages()).isEqualTo(1);
+ dispatcher.removeConsumer(original);
+ assertUnackedMessagesCleared(dispatcher, brokerService, "original removal");
+ }
+ // Use the production subscribe path and the same live connection/protocol identity. No mocked
+ // consumer or manually populated dispatcher collections are needed to create the replacement.
+ Consumer replacement = topic.subscribe(SubscriptionOption.builder()
+ .cnx(original.cnx()).consumerId(original.consumerId()).consumerName(original.consumerName())
+ .subscriptionName(SUBSCRIPTION).subType(original.subType()).isDurable(true)
+ .startMessageId(MessageId.latest).initialPosition(InitialPosition.Latest)
+ .metadata(Collections.emptyMap()).subscriptionProperties(Optional.empty())
+ .keySharedMeta(original.getKeySharedMeta())
+ .build()).get(10, TimeUnit.SECONDS);
+ try {
+ assertThat(replacement).isNotSameAs(original).isEqualTo(original);
+ assertThat(replacement.hashCode()).isEqualTo(original.hashCode());
+ synchronized (dispatcher) {
+ dispatcher.removeConsumer(original);
+ assertThat(dispatcher.getConsumers()).singleElement().isSameAs(replacement);
+ assertUnackedMessagesCleared(dispatcher, brokerService, "stale removal");
+ }
+ // This also checks Key_Shared selector membership, not just the consumer list.
+ replacement.flowPermits(1);
+ assertThat(client.receive(5, TimeUnit.SECONDS)).as("replacement receives replay").isNotNull();
+ synchronized (dispatcher) {
+ assertThat(replacement.getUnackedMessages()).isEqualTo(1);
+ assertThat(dispatcher.getTotalUnackedMessages()).isEqualTo(1);
+ assertThat(brokerService.getTotalUnackedMessages()).isEqualTo(1);
+ }
+ } finally {
+ replacement.close();
+ }
+ assertUnackedMessagesCleared(dispatcher, brokerService, "replacement removal");
+ }
+ }
+
+ @Test(timeOut = 60_000)
+ public void testCloseCleansUpStickyAdditionCompletedAfterRemoval() throws Exception {
+ String topicName = newTopicName();
+ try (PulsarClient otherClient = newPulsarClient(pulsar.getBrokerServiceUrl(), 0);
+ Producer<String> producer = otherClient.newProducer(Schema.STRING)
+ .topic(topicName).enableBatching(false).create();
+ org.apache.pulsar.client.api.Consumer<String> client = pulsarClient.newConsumer(Schema.STRING)
+ .topic(topicName).subscriptionName(SUBSCRIPTION).subscriptionType(SubscriptionType.Key_Shared)
+ .keySharedPolicy(KeySharedPolicy.stickyHashRange().ranges(Range.of(0, 65535))).subscribe()) {
+ PersistentTopic topic =
+ (PersistentTopic) pulsar.getBrokerService().getTopicReference(topicName).orElseThrow();
+ AbstractPersistentDispatcherMultipleConsumers dispatcher =
+ (AbstractPersistentDispatcherMultipleConsumers) topic.getSubscription(SUBSCRIPTION).getDispatcher();
+ StickyKeyConsumerSelector selector = ((StickyKeyDispatcher) dispatcher).getSelector();
+ Consumer original = dispatcher.getConsumers().get(0);
+ ServerCnx originalCnx = (ServerCnx) original.cnx();
+ Channel channel = originalCnx.ctx().channel();
+ // Drain the initial FLOW so it cannot complete the liveness check instead of the held PONG.
+ Awaitility.await().atMost(10, TimeUnit.SECONDS).until(() -> original.getAvailablePermits() > 0);
+ // Hold the real liveness response without blocking the event loop or mocking the selector.
+ channel.eventLoop().submit(() -> channel.config().setAutoRead(false)).get(10, TimeUnit.SECONDS);
+ try {
+ CompletableFuture<Consumer> adding = topic.subscribe(SubscriptionOption.builder()
+ .cnx(topic.getProducers().values().iterator().next().getCnx())
+ .consumerId(1234).consumerName("pending-sticky")
+ .subscriptionName(SUBSCRIPTION).subType(original.subType()).isDurable(true)
+ .startMessageId(MessageId.latest).initialPosition(InitialPosition.Latest)
+ .metadata(Collections.emptyMap()).subscriptionProperties(Optional.empty())
+ .keySharedMeta(original.getKeySharedMeta()).build());
+ // Wait for the liveness request on its owning event loop. The second consumer is already
+ // registered in the dispatcher, but its selector addition must still be outstanding.
+ originalCnx.ctx().executor().submit(() ->
+ assertThat(originalCnx.connectionCheckInProgress).isNotNull().isNotDone())
+ .get(10, TimeUnit.SECONDS);
+ assertThat(adding).isNotDone();
+ assertThat(dispatcher.getConsumers()).hasSize(2);
+ Consumer pending = dispatcher.getConsumers().stream()
+ .filter(c -> c != original).findFirst().orElseThrow();
+ assertThat(pending.cnx()).isNotSameAs(originalCnx);
+ assertThat(selector.select(0)).isSameAs(original);
+
+ // Consumer.close is also used when cursor reset disconnects consumers while subscribe is pending.
+ pending.close();
+ original.close();
+ assertThat(dispatcher.getConsumers()).isEmpty();
+ assertThat(selector.select(0)).isNull();
+ channel.eventLoop().submit(() -> channel.config().setAutoRead(true)).get(10, TimeUnit.SECONDS);
+ assertThat(adding.get(10, TimeUnit.SECONDS)).isSameAs(pending);
+ assertThat(selector.select(0)).as("late addition installed the removed consumer").isSameAs(pending);
+
+ // ServerCnx repeats close when subscribe completes after the client has already timed out.
+ pending.close();
+ assertThat(selector.getConsumerKeyHashRanges()).as("no orphan range after repeated close").isEmpty();
+ try (org.apache.pulsar.client.api.Consumer<String> successor = pulsarClient.newConsumer(Schema.STRING)
+ .topic(topicName).subscriptionName(SUBSCRIPTION).subscriptionType(SubscriptionType.Key_Shared)
+ .keySharedPolicy(KeySharedPolicy.stickyHashRange().ranges(Range.of(0, 65535)))
+ .subscribeAsync().get(10, TimeUnit.SECONDS)) {
+ producer.send("after-cleanup");
+ assertThat(successor.receive(5, TimeUnit.SECONDS)).isNotNull();
+ }
+ } finally {
+ channel.eventLoop().submit(() -> channel.config().setAutoRead(true)).get(10, TimeUnit.SECONDS);
+ }
+ }
+ }
+
private void assertUnackedMessagesCleared(AbstractPersistentDispatcherMultipleConsumers dispatcher,
BrokerService brokerService, String removal) {
assertThat(dispatcher.getTotalUnackedMessages()).as("subscription balance after %s", removal).isZero();