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 2024/10/09 05:38:20 UTC
(pulsar) branch master updated: [improve][broker] PIP-379: Enable the use of the classic implementation of Key_Shared / Shared with feature flag (#23424)
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 676fdb1ffb4 [improve][broker] PIP-379: Enable the use of the classic implementation of Key_Shared / Shared with feature flag (#23424)
676fdb1ffb4 is described below
commit 676fdb1ffb4392bb7b10b8d1e8ba94b379b25166
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Wed Oct 9 08:38:14 2024 +0300
[improve][broker] PIP-379: Enable the use of the classic implementation of Key_Shared / Shared with feature flag (#23424)
---
.../apache/pulsar/broker/ServiceConfiguration.java | 16 +
.../delayed/AbstractDelayedDeliveryTracker.java | 8 +-
.../BucketDelayedDeliveryTrackerFactory.java | 6 +-
.../delayed/DelayedDeliveryTrackerFactory.java | 4 +-
.../delayed/InMemoryDelayedDeliveryTracker.java | 13 +-
.../InMemoryDelayedDeliveryTrackerFactory.java | 6 +-
.../bucket/BucketDelayedDeliveryTracker.java | 6 +-
.../AbstractDispatcherMultipleConsumers.java | 6 -
.../pulsar/broker/service/BrokerService.java | 16 +-
.../org/apache/pulsar/broker/service/Consumer.java | 10 +-
.../apache/pulsar/broker/service/Dispatcher.java | 4 +-
.../pulsar/broker/service/StickyKeyDispatcher.java | 48 ++
...tractPersistentDispatcherMultipleConsumers.java | 67 +++
.../persistent/MessageRedeliveryController.java | 8 +-
.../PersistentDispatcherMultipleConsumers.java | 25 +-
...sistentDispatcherMultipleConsumersClassic.java} | 447 ++++++----------
...istentStickyKeyDispatcherMultipleConsumers.java | 4 +-
...tickyKeyDispatcherMultipleConsumersClassic.java | 583 +++++++++++++++++++++
.../service/persistent/PersistentSubscription.java | 50 +-
.../broker/service/persistent/PersistentTopic.java | 16 +-
.../org/apache/pulsar/broker/BrokerTestUtil.java | 25 +-
.../delayed/AbstractDeliveryTrackerTest.java | 4 +-
.../delayed/DelayedDeliveryTrackerFactoryTest.java | 14 +-
.../delayed/InMemoryDeliveryTrackerTest.java | 7 +-
.../bucket/BucketDelayedDeliveryTrackerTest.java | 4 +-
.../pulsar/broker/service/BatchMessageTest.java | 4 +-
.../BatchMessageWithBatchIndexLevelTest.java | 16 +-
.../PersistentDispatcherFailoverConsumerTest.java | 15 +-
.../pulsar/broker/service/PersistentTopicTest.java | 3 +-
.../persistent/BucketDelayedDeliveryTest.java | 12 +-
.../service/persistent/DelayedDeliveryTest.java | 8 +-
...entDispatcherMultipleConsumersClassicTest.java} | 12 +-
.../PersistentDispatcherMultipleConsumersTest.java | 2 +-
...yKeyDispatcherMultipleConsumersClassicTest.java | 482 +++++++++++++++++
...ntStickyKeyDispatcherMultipleConsumersTest.java | 2 +-
.../pulsar/broker/stats/ConsumerStatsTest.java | 2 +-
.../client/api/KeySharedSubscriptionTest.java | 281 ++++++----
.../SubscriptionMessageDispatchThrottlingTest.java | 18 +-
.../api/SubscriptionPauseOnAckStatPersistTest.java | 5 +-
...ySharedSubscriptionMaxUnackedMessagesTest.java} | 42 +-
.../pulsar/tests/KeySharedImplementationType.java | 61 +++
.../pulsar/common/policies/data/ConsumerStats.java | 4 +-
.../common/policies/data/SubscriptionStats.java | 3 +
.../policies/data/stats/ConsumerStatsImpl.java | 10 +-
.../policies/data/stats/SubscriptionStatsImpl.java | 8 +-
45 files changed, 1827 insertions(+), 560 deletions(-)
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
index 1b021bd5699..19e9ff625ca 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
@@ -817,6 +817,22 @@ public class ServiceConfiguration implements PulsarConfiguration {
+ "The higher the number, the more equal the assignment of keys to consumers")
private int subscriptionKeySharedConsistentHashingReplicaPoints = 100;
+ @FieldContext(
+ category = CATEGORY_POLICIES,
+ doc = "For persistent Key_Shared subscriptions, enables the use of the classic implementation of the "
+ + "Key_Shared subscription that was used before Pulsar 4.0.0 and PIP-379.",
+ dynamic = true
+ )
+ private boolean subscriptionKeySharedUseClassicPersistentImplementation = false;
+
+ @FieldContext(
+ category = CATEGORY_POLICIES,
+ doc = "For persistent Shared subscriptions, enables the use of the classic implementation of the Shared "
+ + "subscription that was used before Pulsar 4.0.0.",
+ dynamic = true
+ )
+ private boolean subscriptionSharedUseClassicPersistentImplementation = false;
+
@FieldContext(
category = CATEGORY_POLICIES,
doc = "Set the default behavior for message deduplication in the broker.\n\n"
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/AbstractDelayedDeliveryTracker.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/AbstractDelayedDeliveryTracker.java
index f93a627bca7..bec5134c4f7 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/AbstractDelayedDeliveryTracker.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/AbstractDelayedDeliveryTracker.java
@@ -24,12 +24,12 @@ import io.netty.util.TimerTask;
import java.time.Clock;
import java.util.concurrent.TimeUnit;
import lombok.extern.slf4j.Slf4j;
-import org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers;
+import org.apache.pulsar.broker.service.persistent.AbstractPersistentDispatcherMultipleConsumers;
@Slf4j
public abstract class AbstractDelayedDeliveryTracker implements DelayedDeliveryTracker, TimerTask {
- protected final PersistentDispatcherMultipleConsumers dispatcher;
+ protected final AbstractPersistentDispatcherMultipleConsumers dispatcher;
// Reference to the shared (per-broker) timer for delayed delivery
protected final Timer timer;
@@ -49,13 +49,13 @@ public abstract class AbstractDelayedDeliveryTracker implements DelayedDeliveryT
private final boolean isDelayedDeliveryDeliverAtTimeStrict;
- public AbstractDelayedDeliveryTracker(PersistentDispatcherMultipleConsumers dispatcher, Timer timer,
+ public AbstractDelayedDeliveryTracker(AbstractPersistentDispatcherMultipleConsumers dispatcher, Timer timer,
long tickTimeMillis,
boolean isDelayedDeliveryDeliverAtTimeStrict) {
this(dispatcher, timer, tickTimeMillis, Clock.systemUTC(), isDelayedDeliveryDeliverAtTimeStrict);
}
- public AbstractDelayedDeliveryTracker(PersistentDispatcherMultipleConsumers dispatcher, Timer timer,
+ public AbstractDelayedDeliveryTracker(AbstractPersistentDispatcherMultipleConsumers dispatcher, Timer timer,
long tickTimeMillis, Clock clock,
boolean isDelayedDeliveryDeliverAtTimeStrict) {
this.dispatcher = dispatcher;
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/BucketDelayedDeliveryTrackerFactory.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/BucketDelayedDeliveryTrackerFactory.java
index 11ad243e0c9..c2d002ad19c 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/BucketDelayedDeliveryTrackerFactory.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/BucketDelayedDeliveryTrackerFactory.java
@@ -36,7 +36,7 @@ import org.apache.pulsar.broker.delayed.bucket.BucketDelayedDeliveryTracker;
import org.apache.pulsar.broker.delayed.bucket.BucketSnapshotStorage;
import org.apache.pulsar.broker.delayed.bucket.RecoverDelayedDeliveryTrackerException;
import org.apache.pulsar.broker.service.BrokerService;
-import org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers;
+import org.apache.pulsar.broker.service.persistent.AbstractPersistentDispatcherMultipleConsumers;
import org.apache.pulsar.common.util.FutureUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -78,7 +78,7 @@ public class BucketDelayedDeliveryTrackerFactory implements DelayedDeliveryTrack
}
@Override
- public DelayedDeliveryTracker newTracker(PersistentDispatcherMultipleConsumers dispatcher) {
+ public DelayedDeliveryTracker newTracker(AbstractPersistentDispatcherMultipleConsumers dispatcher) {
String topicName = dispatcher.getTopic().getName();
String subscriptionName = dispatcher.getSubscription().getName();
BrokerService brokerService = dispatcher.getTopic().getBrokerService();
@@ -97,7 +97,7 @@ public class BucketDelayedDeliveryTrackerFactory implements DelayedDeliveryTrack
}
@VisibleForTesting
- BucketDelayedDeliveryTracker newTracker0(PersistentDispatcherMultipleConsumers dispatcher)
+ BucketDelayedDeliveryTracker newTracker0(AbstractPersistentDispatcherMultipleConsumers dispatcher)
throws RecoverDelayedDeliveryTrackerException {
return new BucketDelayedDeliveryTracker(dispatcher, timer, tickTimeMillis,
isDelayedDeliveryDeliverAtTimeStrict, bucketSnapshotStorage, delayedDeliveryMinIndexCountPerBucket,
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/DelayedDeliveryTrackerFactory.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/DelayedDeliveryTrackerFactory.java
index 5427a46a2e4..763b6d66da1 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/DelayedDeliveryTrackerFactory.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/DelayedDeliveryTrackerFactory.java
@@ -20,7 +20,7 @@ package org.apache.pulsar.broker.delayed;
import com.google.common.annotations.Beta;
import org.apache.pulsar.broker.PulsarService;
-import org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers;
+import org.apache.pulsar.broker.service.persistent.AbstractPersistentDispatcherMultipleConsumers;
/**
* Factory of InMemoryDelayedDeliveryTracker objects. This is the entry point for implementations.
@@ -42,7 +42,7 @@ public interface DelayedDeliveryTrackerFactory extends AutoCloseable {
* @param dispatcher
* a multi-consumer dispatcher instance
*/
- DelayedDeliveryTracker newTracker(PersistentDispatcherMultipleConsumers dispatcher);
+ DelayedDeliveryTracker newTracker(AbstractPersistentDispatcherMultipleConsumers dispatcher);
/**
* Close the factory and release all the resources.
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryDelayedDeliveryTracker.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryDelayedDeliveryTracker.java
index 8bd9fafa137..bdc6e4c814e 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryDelayedDeliveryTracker.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryDelayedDeliveryTracker.java
@@ -28,7 +28,7 @@ import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.PositionFactory;
-import org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers;
+import org.apache.pulsar.broker.service.persistent.AbstractPersistentDispatcherMultipleConsumers;
import org.apache.pulsar.common.util.collections.TripleLongPriorityQueue;
@Slf4j
@@ -52,17 +52,18 @@ public class InMemoryDelayedDeliveryTracker extends AbstractDelayedDeliveryTrack
// Track whether we have seen all messages with fixed delay so far.
private boolean messagesHaveFixedDelay = true;
- InMemoryDelayedDeliveryTracker(PersistentDispatcherMultipleConsumers dispatcher, Timer timer, long tickTimeMillis,
+ InMemoryDelayedDeliveryTracker(AbstractPersistentDispatcherMultipleConsumers dispatcher, Timer timer,
+ long tickTimeMillis,
boolean isDelayedDeliveryDeliverAtTimeStrict,
long fixedDelayDetectionLookahead) {
this(dispatcher, timer, tickTimeMillis, Clock.systemUTC(), isDelayedDeliveryDeliverAtTimeStrict,
fixedDelayDetectionLookahead);
}
- public InMemoryDelayedDeliveryTracker(PersistentDispatcherMultipleConsumers dispatcher, Timer timer,
- long tickTimeMillis, Clock clock,
- boolean isDelayedDeliveryDeliverAtTimeStrict,
- long fixedDelayDetectionLookahead) {
+ public InMemoryDelayedDeliveryTracker(AbstractPersistentDispatcherMultipleConsumers dispatcher, Timer timer,
+ long tickTimeMillis, Clock clock,
+ boolean isDelayedDeliveryDeliverAtTimeStrict,
+ long fixedDelayDetectionLookahead) {
super(dispatcher, timer, tickTimeMillis, clock, isDelayedDeliveryDeliverAtTimeStrict);
this.fixedDelayDetectionLookahead = fixedDelayDetectionLookahead;
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryDelayedDeliveryTrackerFactory.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryDelayedDeliveryTrackerFactory.java
index 179cf74db41..f8b8f5a8ba4 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryDelayedDeliveryTrackerFactory.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryDelayedDeliveryTrackerFactory.java
@@ -25,7 +25,7 @@ import io.netty.util.concurrent.DefaultThreadFactory;
import java.util.concurrent.TimeUnit;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
-import org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers;
+import org.apache.pulsar.broker.service.persistent.AbstractPersistentDispatcherMultipleConsumers;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -51,7 +51,7 @@ public class InMemoryDelayedDeliveryTrackerFactory implements DelayedDeliveryTra
}
@Override
- public DelayedDeliveryTracker newTracker(PersistentDispatcherMultipleConsumers dispatcher) {
+ public DelayedDeliveryTracker newTracker(AbstractPersistentDispatcherMultipleConsumers dispatcher) {
String topicName = dispatcher.getTopic().getName();
String subscriptionName = dispatcher.getSubscription().getName();
DelayedDeliveryTracker tracker = DelayedDeliveryTracker.DISABLE;
@@ -66,7 +66,7 @@ public class InMemoryDelayedDeliveryTrackerFactory implements DelayedDeliveryTra
}
@VisibleForTesting
- InMemoryDelayedDeliveryTracker newTracker0(PersistentDispatcherMultipleConsumers dispatcher) {
+ InMemoryDelayedDeliveryTracker newTracker0(AbstractPersistentDispatcherMultipleConsumers dispatcher) {
return new InMemoryDelayedDeliveryTracker(dispatcher, timer, tickTimeMillis,
isDelayedDeliveryDeliverAtTimeStrict, fixedDelayDetectionLookahead);
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java
index 47c78fa9ee2..0091bf5b9bd 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java
@@ -57,7 +57,7 @@ import org.apache.commons.lang3.tuple.Pair;
import org.apache.pulsar.broker.delayed.AbstractDelayedDeliveryTracker;
import org.apache.pulsar.broker.delayed.proto.DelayedIndex;
import org.apache.pulsar.broker.delayed.proto.SnapshotSegment;
-import org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers;
+import org.apache.pulsar.broker.service.persistent.AbstractPersistentDispatcherMultipleConsumers;
import org.apache.pulsar.common.policies.data.stats.TopicMetricBean;
import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.common.util.collections.TripleLongPriorityQueue;
@@ -105,7 +105,7 @@ public class BucketDelayedDeliveryTracker extends AbstractDelayedDeliveryTracker
private CompletableFuture<Void> pendingLoad = null;
- public BucketDelayedDeliveryTracker(PersistentDispatcherMultipleConsumers dispatcher,
+ public BucketDelayedDeliveryTracker(AbstractPersistentDispatcherMultipleConsumers dispatcher,
Timer timer, long tickTimeMillis,
boolean isDelayedDeliveryDeliverAtTimeStrict,
BucketSnapshotStorage bucketSnapshotStorage,
@@ -117,7 +117,7 @@ public class BucketDelayedDeliveryTracker extends AbstractDelayedDeliveryTracker
maxIndexesPerBucketSnapshotSegment, maxNumBuckets);
}
- public BucketDelayedDeliveryTracker(PersistentDispatcherMultipleConsumers dispatcher,
+ public BucketDelayedDeliveryTracker(AbstractPersistentDispatcherMultipleConsumers dispatcher,
Timer timer, long tickTimeMillis, Clock clock,
boolean isDelayedDeliveryDeliverAtTimeStrict,
BucketSnapshotStorage bucketSnapshotStorage,
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 9fc6b9581a3..e3c2cf40cf3 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
@@ -24,10 +24,7 @@ import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import org.apache.pulsar.broker.ServiceConfiguration;
-import org.apache.pulsar.broker.service.persistent.PersistentStickyKeyDispatcherMultipleConsumers;
import org.apache.pulsar.common.api.proto.CommandSubscribe.SubType;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
/**
*
@@ -239,7 +236,4 @@ public abstract class AbstractDispatcherMultipleConsumers extends AbstractBaseDi
return -1;
}
- private static final Logger log = LoggerFactory.getLogger(PersistentStickyKeyDispatcherMultipleConsumers.class);
-
-
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
index 8d0b9a4a84e..fee5e25647c 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
@@ -129,8 +129,8 @@ import org.apache.pulsar.broker.service.BrokerServiceException.TopicMigratedExce
import org.apache.pulsar.broker.service.TopicEventsListener.EventStage;
import org.apache.pulsar.broker.service.TopicEventsListener.TopicEvent;
import org.apache.pulsar.broker.service.nonpersistent.NonPersistentTopic;
+import org.apache.pulsar.broker.service.persistent.AbstractPersistentDispatcherMultipleConsumers;
import org.apache.pulsar.broker.service.persistent.DispatchRateLimiter;
-import org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.broker.service.persistent.SystemTopic;
import org.apache.pulsar.broker.service.plugin.EntryFilterProvider;
@@ -301,7 +301,7 @@ public class BrokerService implements Closeable {
private final int maxUnackedMessages;
public final int maxUnackedMsgsPerDispatcher;
private final AtomicBoolean blockedDispatcherOnHighUnackedMsgs = new AtomicBoolean(false);
- private final Set<PersistentDispatcherMultipleConsumers> blockedDispatchers = ConcurrentHashMap.newKeySet();
+ private final Set<AbstractPersistentDispatcherMultipleConsumers> blockedDispatchers = ConcurrentHashMap.newKeySet();
private final ReadWriteLock lock = new ReentrantReadWriteLock();
@VisibleForTesting
private final DelayedDeliveryTrackerFactory delayedDeliveryTrackerFactory;
@@ -3328,7 +3328,7 @@ public class BrokerService implements Closeable {
* @param dispatcher
* @param numberOfMessages
*/
- public void addUnAckedMessages(PersistentDispatcherMultipleConsumers dispatcher, int numberOfMessages) {
+ public void addUnAckedMessages(AbstractPersistentDispatcherMultipleConsumers dispatcher, int numberOfMessages) {
// don't block dispatchers if maxUnackedMessages = 0
if (maxUnackedMessages > 0) {
totalUnackedMessages.add(numberOfMessages);
@@ -3387,10 +3387,10 @@ public class BrokerService implements Closeable {
try {
forEachTopic(topic -> {
topic.getSubscriptions().forEach((subName, persistentSubscription) -> {
- if (persistentSubscription.getDispatcher() instanceof PersistentDispatcherMultipleConsumers) {
- PersistentDispatcherMultipleConsumers dispatcher =
- (PersistentDispatcherMultipleConsumers) persistentSubscription
- .getDispatcher();
+ if (persistentSubscription.getDispatcher()
+ instanceof AbstractPersistentDispatcherMultipleConsumers) {
+ AbstractPersistentDispatcherMultipleConsumers dispatcher =
+ (AbstractPersistentDispatcherMultipleConsumers) persistentSubscription.getDispatcher();
int dispatcherUnAckMsgs = dispatcher.getTotalUnackedMessages();
if (dispatcherUnAckMsgs > maxUnackedMsgsPerDispatcher) {
log.info("[{}] Blocking dispatcher due to reached max broker limit {}",
@@ -3411,7 +3411,7 @@ public class BrokerService implements Closeable {
*
* @param dispatcherList
*/
- public void unblockDispatchersOnUnAckMessages(List<PersistentDispatcherMultipleConsumers> dispatcherList) {
+ public void unblockDispatchersOnUnAckMessages(List<AbstractPersistentDispatcherMultipleConsumers> dispatcherList) {
lock.writeLock().lock();
try {
dispatcherList.forEach(dispatcher -> {
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 c9584f2c179..d25ebd0839d 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
@@ -146,7 +146,7 @@ public class Consumer {
private static final double avgPercent = 0.9;
private boolean preciseDispatcherFlowControl;
- private Position lastSentPositionWhenJoining;
+ private Position readPositionWhenJoining;
private final String clientAddress; // IP address only, no port number included
private final MessageId startMessageId;
private final boolean isAcknowledgmentAtBatchIndexLevelEnabled;
@@ -973,8 +973,8 @@ public class Consumer {
stats.unackedMessages = unackedMessages;
stats.blockedConsumerOnUnackedMsgs = blockedConsumerOnUnackedMsgs;
stats.avgMessagesPerEntry = getAvgMessagesPerEntry();
- if (lastSentPositionWhenJoining != null) {
- stats.lastSentPositionWhenJoining = lastSentPositionWhenJoining.toString();
+ if (readPositionWhenJoining != null) {
+ stats.readPositionWhenJoining = readPositionWhenJoining.toString();
}
return stats;
}
@@ -1189,8 +1189,8 @@ public class Consumer {
return preciseDispatcherFlowControl;
}
- public void setLastSentPositionWhenJoining(Position lastSentPositionWhenJoining) {
- this.lastSentPositionWhenJoining = lastSentPositionWhenJoining;
+ public void setReadPositionWhenJoining(Position readPositionWhenJoining) {
+ this.readPositionWhenJoining = readPositionWhenJoining;
}
public int getMaxUnackedMessages() {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Dispatcher.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Dispatcher.java
index d1d44709a9c..f68a9a0986b 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Dispatcher.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Dispatcher.java
@@ -153,8 +153,8 @@ public interface Dispatcher {
/**
* Trigger a new "readMoreEntries" if the dispatching has been paused before. This method is only implemented in
- * {@link org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers} right now, other
- * implements are not necessary to implement this method.
+ * {@link org.apache.pulsar.broker.service.persistent.AbstractPersistentDispatcherMultipleConsumers} right now,
+ * other implementations do not necessary implement this method.
* @return did a resume.
*/
default boolean checkAndResumeIfPaused(){
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/StickyKeyDispatcher.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/StickyKeyDispatcher.java
new file mode 100644
index 00000000000..79ce3081584
--- /dev/null
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/StickyKeyDispatcher.java
@@ -0,0 +1,48 @@
+/*
+ * 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.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.pulsar.client.api.Range;
+import org.apache.pulsar.common.api.proto.KeySharedMeta;
+import org.apache.pulsar.common.api.proto.KeySharedMode;
+
+public interface StickyKeyDispatcher extends Dispatcher {
+
+ boolean hasSameKeySharedPolicy(KeySharedMeta ksm);
+
+ Map<Consumer, List<Range>> getConsumerKeyHashRanges();
+
+ boolean isAllowOutOfOrderDelivery();
+
+ KeySharedMode getKeySharedMode();
+
+ StickyKeyConsumerSelector getSelector();
+
+ long getNumberOfMessagesInReplay();
+
+ default LinkedHashMap<Consumer, Position> getRecentlyJoinedConsumers() {
+ return null;
+ }
+
+ boolean isClassic();
+}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/AbstractPersistentDispatcherMultipleConsumers.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/AbstractPersistentDispatcherMultipleConsumers.java
new file mode 100644
index 00000000000..79d365b9fee
--- /dev/null
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/AbstractPersistentDispatcherMultipleConsumers.java
@@ -0,0 +1,67 @@
+/*
+ * 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 java.util.Map;
+import org.apache.bookkeeper.mledger.AsyncCallbacks;
+import org.apache.bookkeeper.mledger.ManagedCursor;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.service.AbstractDispatcherMultipleConsumers;
+import org.apache.pulsar.broker.service.Dispatcher;
+import org.apache.pulsar.broker.service.Subscription;
+import org.apache.pulsar.broker.service.Topic;
+import org.apache.pulsar.common.policies.data.stats.TopicMetricBean;
+
+public abstract class AbstractPersistentDispatcherMultipleConsumers extends AbstractDispatcherMultipleConsumers
+ implements Dispatcher, AsyncCallbacks.ReadEntriesCallback {
+ public AbstractPersistentDispatcherMultipleConsumers(Subscription subscription,
+ ServiceConfiguration serviceConfig) {
+ super(subscription, serviceConfig);
+ }
+
+ public abstract void unBlockDispatcherOnUnackedMsgs();
+
+ public abstract void readMoreEntriesAsync();
+
+ public abstract String getName();
+
+ public abstract boolean isBlockedDispatcherOnUnackedMsgs();
+
+ public abstract int getTotalUnackedMessages();
+
+ public abstract void blockDispatcherOnUnackedMsgs();
+
+ public abstract long getNumberOfMessagesInReplay();
+
+ public abstract boolean isHavePendingRead();
+
+ public abstract boolean isHavePendingReplayRead();
+
+ public abstract ManagedCursor getCursor();
+
+ public abstract Topic getTopic();
+
+ public abstract Subscription getSubscription();
+
+ public abstract long getDelayedTrackerMemoryUsage();
+
+ public abstract Map<String, TopicMetricBean> getBucketDelayedIndexStats();
+
+ public abstract boolean isClassic();
+}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageRedeliveryController.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageRedeliveryController.java
index b34a0b45438..46f1f0a5356 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageRedeliveryController.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageRedeliveryController.java
@@ -41,12 +41,18 @@ import org.apache.pulsar.utils.ConcurrentBitmapSortedLongPairSet;
public class MessageRedeliveryController {
private final boolean allowOutOfOrderDelivery;
+ private final boolean isClassicDispatcher;
private final ConcurrentBitmapSortedLongPairSet messagesToRedeliver;
private final ConcurrentLongLongPairHashMap hashesToBeBlocked;
private final ConcurrentLongLongHashMap hashesRefCount;
public MessageRedeliveryController(boolean allowOutOfOrderDelivery) {
+ this(allowOutOfOrderDelivery, false);
+ }
+
+ public MessageRedeliveryController(boolean allowOutOfOrderDelivery, boolean isClassicDispatcher) {
this.allowOutOfOrderDelivery = allowOutOfOrderDelivery;
+ this.isClassicDispatcher = isClassicDispatcher;
this.messagesToRedeliver = new ConcurrentBitmapSortedLongPairSet();
if (!allowOutOfOrderDelivery) {
this.hashesToBeBlocked = ConcurrentLongLongPairHashMap
@@ -65,7 +71,7 @@ public class MessageRedeliveryController {
public void add(long ledgerId, long entryId, long stickyKeyHash) {
if (!allowOutOfOrderDelivery) {
- if (stickyKeyHash == STICKY_KEY_HASH_NOT_SET) {
+ if (!isClassicDispatcher && stickyKeyHash == STICKY_KEY_HASH_NOT_SET) {
throw new IllegalArgumentException("Sticky key hash is not set. It is required.");
}
boolean inserted = hashesToBeBlocked.putIfAbsent(ledgerId, entryId, stickyKeyHash, 0);
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 73d152bab1a..b1cd186c317 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
@@ -39,7 +39,6 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.function.Predicate;
-import org.apache.bookkeeper.mledger.AsyncCallbacks.ReadEntriesCallback;
import org.apache.bookkeeper.mledger.Entry;
import org.apache.bookkeeper.mledger.ManagedCursor;
import org.apache.bookkeeper.mledger.ManagedLedgerException;
@@ -60,7 +59,6 @@ import org.apache.pulsar.broker.service.AbstractDispatcherMultipleConsumers;
import org.apache.pulsar.broker.service.BrokerServiceException;
import org.apache.pulsar.broker.service.BrokerServiceException.ConsumerBusyException;
import org.apache.pulsar.broker.service.Consumer;
-import org.apache.pulsar.broker.service.Dispatcher;
import org.apache.pulsar.broker.service.EntryAndMetadata;
import org.apache.pulsar.broker.service.EntryBatchIndexesAcks;
import org.apache.pulsar.broker.service.EntryBatchSizes;
@@ -85,8 +83,7 @@ import org.slf4j.LoggerFactory;
/**
*
*/
-public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMultipleConsumers
- implements Dispatcher, ReadEntriesCallback {
+public class PersistentDispatcherMultipleConsumers extends AbstractPersistentDispatcherMultipleConsumers {
protected final PersistentTopic topic;
protected final ManagedCursor cursor;
protected volatile Range<Position> lastIndividualDeletedRangeFromCursorRecovery;
@@ -162,7 +159,7 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
this.name = topic.getName() + " / " + Codec.decode(cursor.getName());
this.topic = topic;
this.dispatchMessagesThread = topic.getBrokerService().getTopicOrderedExecutor().chooseThread();
- this.redeliveryMessages = new MessageRedeliveryController(allowOutOfOrderDelivery);
+ this.redeliveryMessages = new MessageRedeliveryController(allowOutOfOrderDelivery, false);
this.redeliveryTracker = this.serviceConfig.isSubscriptionRedeliveryTrackerEnabled()
? new InMemoryRedeliveryTracker()
: RedeliveryTrackerDisabled.REDELIVERY_TRACKER_DISABLED;
@@ -320,6 +317,7 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
* We should not call readMoreEntries() recursively in the same thread as there is a risk of StackOverflowError.
*
*/
+ @Override
public void readMoreEntriesAsync() {
// deduplication for readMoreEntriesAsync calls
if (readMoreEntriesAsyncRequested.compareAndSet(false, true)) {
@@ -1285,6 +1283,7 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
blockedDispatcherOnUnackedMsgs = TRUE;
}
+ @Override
public void unBlockDispatcherOnUnackedMsgs() {
blockedDispatcherOnUnackedMsgs = FALSE;
}
@@ -1293,6 +1292,7 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
return totalUnackedMessages;
}
+ @Override
public String getName() {
return name;
}
@@ -1488,6 +1488,11 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
return Collections.emptyMap();
}
+ @Override
+ public boolean isClassic() {
+ return false;
+ }
+
public ManagedCursor getCursor() {
return cursor;
}
@@ -1505,5 +1510,15 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
return redeliveryMessages.size();
}
+ @Override
+ public boolean isHavePendingRead() {
+ return havePendingRead;
+ }
+
+ @Override
+ public boolean isHavePendingReplayRead() {
+ return havePendingReplayRead;
+ }
+
private static final Logger log = LoggerFactory.getLogger(PersistentDispatcherMultipleConsumers.class);
}
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/PersistentDispatcherMultipleConsumersClassic.java
similarity index 79%
copy from pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java
copy to pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumersClassic.java
index 73d152bab1a..6ab7acfa56d 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/PersistentDispatcherMultipleConsumersClassic.java
@@ -18,7 +18,6 @@
*/
package org.apache.pulsar.broker.service.persistent;
-import static org.apache.pulsar.broker.service.StickyKeyConsumerSelector.STICKY_KEY_HASH_NOT_SET;
import static org.apache.pulsar.broker.service.persistent.PersistentTopic.MESSAGE_RATE_BACKOFF_MS;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Lists;
@@ -39,7 +38,6 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.function.Predicate;
-import org.apache.bookkeeper.mledger.AsyncCallbacks.ReadEntriesCallback;
import org.apache.bookkeeper.mledger.Entry;
import org.apache.bookkeeper.mledger.ManagedCursor;
import org.apache.bookkeeper.mledger.ManagedLedgerException;
@@ -47,9 +45,7 @@ import org.apache.bookkeeper.mledger.ManagedLedgerException.NoMoreEntriesToReadE
import org.apache.bookkeeper.mledger.ManagedLedgerException.TooManyRequestsException;
import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.PositionFactory;
-import org.apache.commons.lang3.mutable.MutableBoolean;
import org.apache.commons.lang3.tuple.Pair;
-import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.delayed.BucketDelayedDeliveryTrackerFactory;
import org.apache.pulsar.broker.delayed.DelayedDeliveryTracker;
import org.apache.pulsar.broker.delayed.DelayedDeliveryTrackerFactory;
@@ -60,7 +56,6 @@ import org.apache.pulsar.broker.service.AbstractDispatcherMultipleConsumers;
import org.apache.pulsar.broker.service.BrokerServiceException;
import org.apache.pulsar.broker.service.BrokerServiceException.ConsumerBusyException;
import org.apache.pulsar.broker.service.Consumer;
-import org.apache.pulsar.broker.service.Dispatcher;
import org.apache.pulsar.broker.service.EntryAndMetadata;
import org.apache.pulsar.broker.service.EntryBatchIndexesAcks;
import org.apache.pulsar.broker.service.EntryBatchSizes;
@@ -69,6 +64,7 @@ import org.apache.pulsar.broker.service.RedeliveryTracker;
import org.apache.pulsar.broker.service.RedeliveryTrackerDisabled;
import org.apache.pulsar.broker.service.SendMessageInfo;
import org.apache.pulsar.broker.service.SharedConsumerAssignor;
+import org.apache.pulsar.broker.service.StickyKeyConsumerSelector;
import org.apache.pulsar.broker.service.Subscription;
import org.apache.pulsar.broker.service.persistent.DispatchRateLimiter.Type;
import org.apache.pulsar.broker.transaction.exception.buffer.TransactionBufferException;
@@ -83,10 +79,12 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
- *
+ * This is the "classic" dispatcher implementation for the Shared subscription that was used before
+ * Pulsar 4.0.0 and PIP-379. When `subscriptionSharedUseClassicPersistentImplementation=true`,
+ * this dispatcher will be used. The main purpose is to provide a way to rollback to the old behavior
+ * in case of issues with the preferred implementation.
*/
-public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMultipleConsumers
- implements Dispatcher, ReadEntriesCallback {
+public class PersistentDispatcherMultipleConsumersClassic extends AbstractPersistentDispatcherMultipleConsumers {
protected final PersistentTopic topic;
protected final ManagedCursor cursor;
protected volatile Range<Position> lastIndividualDeletedRangeFromCursorRecovery;
@@ -103,16 +101,16 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
protected boolean shouldRewindBeforeReadingOrReplaying = false;
protected final String name;
private boolean sendInProgress = false;
- protected static final AtomicIntegerFieldUpdater<PersistentDispatcherMultipleConsumers>
+ protected static final AtomicIntegerFieldUpdater<PersistentDispatcherMultipleConsumersClassic>
TOTAL_AVAILABLE_PERMITS_UPDATER =
- AtomicIntegerFieldUpdater.newUpdater(PersistentDispatcherMultipleConsumers.class,
+ AtomicIntegerFieldUpdater.newUpdater(PersistentDispatcherMultipleConsumersClassic.class,
"totalAvailablePermits");
protected volatile int totalAvailablePermits = 0;
protected volatile int readBatchSize;
protected final Backoff readFailureBackoff;
- private static final AtomicIntegerFieldUpdater<PersistentDispatcherMultipleConsumers>
+ private static final AtomicIntegerFieldUpdater<PersistentDispatcherMultipleConsumersClassic>
TOTAL_UNACKED_MESSAGES_UPDATER =
- AtomicIntegerFieldUpdater.newUpdater(PersistentDispatcherMultipleConsumers.class,
+ AtomicIntegerFieldUpdater.newUpdater(PersistentDispatcherMultipleConsumersClassic.class,
"totalUnackedMessages");
protected volatile int totalUnackedMessages = 0;
/**
@@ -121,63 +119,50 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
* too many CPU circles, see {@link #afterAckMessages(Throwable, Object)} for more details. Do not use this
* to confirm whether the delivery should be paused, please call {@link #shouldPauseOnAckStatePersist}.
*/
- protected static final AtomicIntegerFieldUpdater<PersistentDispatcherMultipleConsumers>
+ protected static final AtomicIntegerFieldUpdater<PersistentDispatcherMultipleConsumersClassic>
BLOCKED_DISPATCHER_ON_CURSOR_DATA_CAN_NOT_FULLY_PERSIST_UPDATER =
- AtomicIntegerFieldUpdater.newUpdater(PersistentDispatcherMultipleConsumers.class,
+ AtomicIntegerFieldUpdater.newUpdater(PersistentDispatcherMultipleConsumersClassic.class,
"blockedDispatcherOnCursorDataCanNotFullyPersist");
private volatile int blockedDispatcherOnCursorDataCanNotFullyPersist = FALSE;
private volatile int blockedDispatcherOnUnackedMsgs = FALSE;
- protected static final AtomicIntegerFieldUpdater<PersistentDispatcherMultipleConsumers>
+ protected static final AtomicIntegerFieldUpdater<PersistentDispatcherMultipleConsumersClassic>
BLOCKED_DISPATCHER_ON_UNACKMSG_UPDATER =
- AtomicIntegerFieldUpdater.newUpdater(PersistentDispatcherMultipleConsumers.class,
+ AtomicIntegerFieldUpdater.newUpdater(PersistentDispatcherMultipleConsumersClassic.class,
"blockedDispatcherOnUnackedMsgs");
protected Optional<DispatchRateLimiter> dispatchRateLimiter = Optional.empty();
private AtomicBoolean isRescheduleReadInProgress = new AtomicBoolean(false);
- private final AtomicBoolean readMoreEntriesAsyncRequested = new AtomicBoolean(false);
protected final ExecutorService dispatchMessagesThread;
private final SharedConsumerAssignor assignor;
- // tracks how many entries were processed by consumers in the last trySendMessagesToConsumers call
- // the number includes also delayed messages, marker messages, aborted txn messages and filtered messages
- // When no messages were processed, the value is 0. This is also an indication that the dispatcher didn't
- // make progress in the last trySendMessagesToConsumers call.
- protected int lastNumberOfEntriesProcessed;
- protected boolean skipNextBackoff;
- private final Backoff retryBackoff;
+
protected enum ReadType {
Normal, Replay
}
- private Position lastMarkDeletePositionBeforeReadMoreEntries;
- private volatile long readMoreEntriesCallCount;
- public PersistentDispatcherMultipleConsumers(PersistentTopic topic, ManagedCursor cursor,
- Subscription subscription) {
+ public PersistentDispatcherMultipleConsumersClassic(PersistentTopic topic, ManagedCursor cursor,
+ Subscription subscription) {
this(topic, cursor, subscription, true);
}
- public PersistentDispatcherMultipleConsumers(PersistentTopic topic, ManagedCursor cursor, Subscription subscription,
- boolean allowOutOfOrderDelivery) {
+ public PersistentDispatcherMultipleConsumersClassic(PersistentTopic topic, ManagedCursor cursor,
+ Subscription subscription,
+ boolean allowOutOfOrderDelivery) {
super(subscription, topic.getBrokerService().pulsar().getConfiguration());
this.cursor = cursor;
this.lastIndividualDeletedRangeFromCursorRecovery = cursor.getLastIndividualDeletedRange();
this.name = topic.getName() + " / " + Codec.decode(cursor.getName());
this.topic = topic;
this.dispatchMessagesThread = topic.getBrokerService().getTopicOrderedExecutor().chooseThread();
- this.redeliveryMessages = new MessageRedeliveryController(allowOutOfOrderDelivery);
+ this.redeliveryMessages = new MessageRedeliveryController(allowOutOfOrderDelivery, true);
this.redeliveryTracker = this.serviceConfig.isSubscriptionRedeliveryTrackerEnabled()
? new InMemoryRedeliveryTracker()
: RedeliveryTrackerDisabled.REDELIVERY_TRACKER_DISABLED;
this.readBatchSize = serviceConfig.getDispatcherMaxReadBatchSize();
this.initializeDispatchRateLimiterIfNeeded();
- this.assignor = new SharedConsumerAssignor(this::getNextConsumer, this::addEntryToReplay);
- ServiceConfiguration serviceConfiguration = topic.getBrokerService().pulsar().getConfiguration();
+ this.assignor = new SharedConsumerAssignor(this::getNextConsumer, this::addMessageToReplay);
this.readFailureBackoff = new Backoff(
- serviceConfiguration.getDispatcherReadFailureBackoffInitialTimeInMs(),
+ topic.getBrokerService().pulsar().getConfiguration().getDispatcherReadFailureBackoffInitialTimeInMs(),
TimeUnit.MILLISECONDS,
1, TimeUnit.MINUTES, 0, TimeUnit.MILLISECONDS);
- retryBackoff = new Backoff(
- serviceConfiguration.getDispatcherRetryBackoffInitialTimeInMs(), TimeUnit.MILLISECONDS,
- serviceConfiguration.getDispatcherRetryBackoffMaxTimeInMs(), TimeUnit.MILLISECONDS,
- 0, TimeUnit.MILLISECONDS);
}
@Override
@@ -243,12 +228,8 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
if (log.isDebugEnabled()) {
log.debug("[{}] Consumer are left, reading more entries", name);
}
- MutableBoolean notifyAddedToReplay = new MutableBoolean(false);
- consumer.getPendingAcks().forEachAndClose((ledgerId, entryId, batchSize, stickyKeyHash) -> {
- boolean addedToReplay = addMessageToReplay(ledgerId, entryId, stickyKeyHash);
- if (addedToReplay) {
- notifyAddedToReplay.setTrue();
- }
+ consumer.getPendingAcks().forEach((ledgerId, entryId, batchSize, stickyKeyHash) -> {
+ addMessageToReplay(ledgerId, entryId, stickyKeyHash);
});
totalAvailablePermits -= consumer.getAvailablePermits();
if (log.isDebugEnabled()) {
@@ -256,9 +237,7 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
+ "New dispatcher permit count is {}", name, consumer.getAvailablePermits(),
totalAvailablePermits);
}
- if (notifyAddedToReplay.booleanValue()) {
- notifyRedeliveryMessageAdded();
- }
+ readMoreEntries();
}
} else {
/**
@@ -274,12 +253,7 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
}
}
- protected synchronized void internalRemoveConsumer(Consumer consumer) {
- consumerSet.removeAll(consumer);
- consumerList.remove(consumer);
- }
-
- protected synchronized void clearComponentsAfterRemovedAllConsumers() {
+ private synchronized void clearComponentsAfterRemovedAllConsumers() {
cancelPendingRead();
redeliveryMessages.clear();
@@ -313,7 +287,7 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
+ "after adding {} permits", name, consumer,
totalAvailablePermits, additionalNumberOfMessages);
}
- readMoreEntriesAsync();
+ readMoreEntries();
}
/**
@@ -321,13 +295,7 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
*
*/
public void readMoreEntriesAsync() {
- // deduplication for readMoreEntriesAsync calls
- if (readMoreEntriesAsyncRequested.compareAndSet(false, true)) {
- topic.getBrokerService().executor().execute(() -> {
- readMoreEntriesAsyncRequested.set(false);
- readMoreEntries();
- });
- }
+ topic.getBrokerService().executor().execute(this::readMoreEntries);
}
public synchronized void readMoreEntries() {
@@ -358,20 +326,6 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
return;
}
- // increment the counter for readMoreEntries calls, to track the number of times readMoreEntries is called
- readMoreEntriesCallCount++;
-
- // remove possible expired messages from redelivery tracker and pending acks
- Position markDeletePosition = cursor.getMarkDeletedPosition();
- if (lastMarkDeletePositionBeforeReadMoreEntries != markDeletePosition) {
- redeliveryMessages.removeAllUpTo(markDeletePosition.getLedgerId(), markDeletePosition.getEntryId());
- for (Consumer consumer : consumerList) {
- consumer.getPendingAcks()
- .removeAllUpTo(markDeletePosition.getLedgerId(), markDeletePosition.getEntryId());
- }
- lastMarkDeletePositionBeforeReadMoreEntries = markDeletePosition;
- }
-
// totalAvailablePermits may be updated by other threads
int firstAvailableConsumerPermits = getFirstAvailableConsumerPermits();
int currentTotalAvailablePermits = Math.max(totalAvailablePermits, firstAvailableConsumerPermits);
@@ -385,24 +339,26 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
return;
}
- Set<Position> messagesToReplayNow =
- canReplayMessages() ? getMessagesToReplayNow(messagesToRead) : Collections.emptySet();
- if (!messagesToReplayNow.isEmpty()) {
+ NavigableSet<Position> messagesToReplayNow = getMessagesToReplayNow(messagesToRead);
+ NavigableSet<Position> messagesToReplayFiltered = filterOutEntriesWillBeDiscarded(messagesToReplayNow);
+ if (!messagesToReplayFiltered.isEmpty()) {
if (log.isDebugEnabled()) {
log.debug("[{}] Schedule replay of {} messages for {} consumers", name,
- messagesToReplayNow.size(), consumerList.size());
+ messagesToReplayFiltered.size(), consumerList.size());
}
+
havePendingReplayRead = true;
- updateMinReplayedPosition();
+ minReplayedPosition = messagesToReplayNow.first();
Set<? extends Position> deletedMessages = topic.isDelayedDeliveryEnabled()
- ? asyncReplayEntriesInOrder(messagesToReplayNow)
- : asyncReplayEntries(messagesToReplayNow);
+ ? asyncReplayEntriesInOrder(messagesToReplayFiltered)
+ : asyncReplayEntries(messagesToReplayFiltered);
// clear already acked positions from replay bucket
- deletedMessages.forEach(position -> redeliveryMessages.remove(position.getLedgerId(),
- position.getEntryId()));
+
+ deletedMessages.forEach(position -> redeliveryMessages.remove(((Position) position).getLedgerId(),
+ ((Position) position).getEntryId()));
// if all the entries are acked-entries and cleared up from redeliveryMessages, try to read
// next entries as readCompletedEntries-callback was never called
- if ((messagesToReplayNow.size() - deletedMessages.size()) == 0) {
+ if ((messagesToReplayFiltered.size() - deletedMessages.size()) == 0) {
havePendingReplayRead = false;
readMoreEntriesAsync();
}
@@ -411,11 +367,7 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
log.debug("[{}] Dispatcher read is blocked due to unackMessages {} reached to max {}", name,
totalUnackedMessages, topic.getMaxUnackedMessagesOnSubscription());
}
- } else if (doesntHavePendingRead()) {
- if (!isNormalReadAllowed()) {
- handleNormalReadNotAllowed();
- return;
- }
+ } else if (!havePendingRead && hasConsumersNeededNormalRead()) {
if (shouldPauseOnAckStatePersist(ReadType.Normal)) {
if (log.isDebugEnabled()) {
log.debug("[{}] [{}] Skipping read for the topic, Due to blocked on ack state persistent.",
@@ -428,14 +380,38 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
consumerList.size());
}
havePendingRead = true;
- updateMinReplayedPosition();
+ NavigableSet<Position> toReplay = getMessagesToReplayNow(1);
+ if (!toReplay.isEmpty()) {
+ minReplayedPosition = toReplay.first();
+ redeliveryMessages.add(minReplayedPosition.getLedgerId(), minReplayedPosition.getEntryId());
+ } else {
+ minReplayedPosition = null;
+ }
- messagesToRead = Math.min(messagesToRead, getMaxEntriesReadLimit());
- cursor.asyncReadEntriesWithSkipOrWait(messagesToRead, bytesToRead, this, ReadType.Normal,
- topic.getMaxReadPosition(), createReadEntriesSkipConditionForNormalRead());
+ // Filter out and skip read delayed messages exist in DelayedDeliveryTracker
+ if (delayedDeliveryTracker.isPresent()) {
+ Predicate<Position> skipCondition = null;
+ final DelayedDeliveryTracker deliveryTracker = delayedDeliveryTracker.get();
+ if (deliveryTracker instanceof BucketDelayedDeliveryTracker) {
+ skipCondition = position -> ((BucketDelayedDeliveryTracker) deliveryTracker)
+ .containsMessage(position.getLedgerId(), position.getEntryId());
+ }
+ cursor.asyncReadEntriesWithSkipOrWait(messagesToRead, bytesToRead, this, ReadType.Normal,
+ topic.getMaxReadPosition(), skipCondition);
+ } else {
+ cursor.asyncReadEntriesOrWait(messagesToRead, bytesToRead, this, ReadType.Normal,
+ topic.getMaxReadPosition());
+ }
} else {
if (log.isDebugEnabled()) {
- log.debug("[{}] Cannot schedule next read until previous one is done", name);
+ if (!messagesToReplayNow.isEmpty()) {
+ log.debug("[{}] [{}] Skipping read for the topic: because all entries in replay queue were"
+ + " filtered out due to the mechanism of Key_Shared mode, and the left consumers have"
+ + " no permits now",
+ topic.getName(), getSubscriptionName());
+ } else {
+ log.debug("[{}] Cannot schedule next read until previous one is done", name);
+ }
}
}
} else {
@@ -445,60 +421,6 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
}
}
- protected Predicate<Position> createReadEntriesSkipConditionForNormalRead() {
- Predicate<Position> skipCondition = null;
- // Filter out and skip read delayed messages exist in DelayedDeliveryTracker
- if (delayedDeliveryTracker.isPresent()) {
- final DelayedDeliveryTracker deliveryTracker = delayedDeliveryTracker.get();
- if (deliveryTracker instanceof BucketDelayedDeliveryTracker) {
- skipCondition = position -> ((BucketDelayedDeliveryTracker) deliveryTracker)
- .containsMessage(position.getLedgerId(), position.getEntryId());
- }
- }
- return skipCondition;
- }
-
- /**
- * Sets a hard limit on the number of entries to read from the Managed Ledger.
- * Subclasses can override this method to set a different limit.
- * By default, this method does not impose an additional limit.
- *
- * @return the maximum number of entries to read from the Managed Ledger
- */
- protected int getMaxEntriesReadLimit() {
- return Integer.MAX_VALUE;
- }
-
- /**
- * Checks if there's a pending read operation that hasn't completed yet.
- * This allows to avoid scheduling a new read operation while the previous one is still in progress.
- * @return true if there's a pending read operation
- */
- protected boolean doesntHavePendingRead() {
- return !havePendingRead;
- }
-
- protected void handleNormalReadNotAllowed() {
- // do nothing
- }
-
- protected long getReadMoreEntriesCallCount() {
- return readMoreEntriesCallCount;
- }
-
- /**
- * Controls whether replaying entries is currently enabled.
- * Subclasses can override this method to temporarily disable replaying entries.
- * @return true if replaying entries is currently enabled
- */
- protected boolean canReplayMessages() {
- return true;
- }
-
- private void updateMinReplayedPosition() {
- minReplayedPosition = getFirstPositionInReplay().orElse(null);
- }
-
private boolean shouldPauseOnAckStatePersist(ReadType readType) {
// Allows new consumers to consume redelivered messages caused by the just-closed consumer.
if (readType != ReadType.Normal) {
@@ -515,27 +437,16 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
@Override
protected void reScheduleRead() {
- reScheduleReadInMs(MESSAGE_RATE_BACKOFF_MS);
- }
-
- protected synchronized void reScheduleReadWithBackoff() {
- reScheduleReadInMs(retryBackoff.next());
- }
-
- protected void reScheduleReadInMs(long readAfterMs) {
if (isRescheduleReadInProgress.compareAndSet(false, true)) {
if (log.isDebugEnabled()) {
- log.debug("[{}] [{}] Reschedule message read in {} ms", topic.getName(), name, readAfterMs);
- }
- Runnable runnable = () -> {
- isRescheduleReadInProgress.set(false);
- readMoreEntries();
- };
- if (readAfterMs > 0) {
- topic.getBrokerService().executor().schedule(runnable, readAfterMs, TimeUnit.MILLISECONDS);
- } else {
- topic.getBrokerService().executor().execute(runnable);
+ log.debug("[{}] [{}] Reschedule message read in {} ms", topic.getName(), name, MESSAGE_RATE_BACKOFF_MS);
}
+ topic.getBrokerService().executor().schedule(
+ () -> {
+ isRescheduleReadInProgress.set(false);
+ readMoreEntries();
+ },
+ MESSAGE_RATE_BACKOFF_MS, TimeUnit.MILLISECONDS);
}
}
@@ -741,7 +652,7 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
entries.forEach(Entry::release);
cursor.rewind();
shouldRewindBeforeReadingOrReplaying = false;
- readMoreEntriesAsync();
+ readMoreEntries();
return;
}
@@ -749,8 +660,8 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
log.debug("[{}] Distributing {} messages to {} consumers", name, entries.size(), consumerList.size());
}
- long totalBytesSize = entries.stream().mapToLong(Entry::getLength).sum();
- updatePendingBytesToDispatch(totalBytesSize);
+ long size = entries.stream().mapToLong(Entry::getLength).sum();
+ updatePendingBytesToDispatch(size);
// dispatch messages to a separate thread, but still in order for this subscription
// sendMessagesToConsumers is responsible for running broker-side filters
@@ -760,34 +671,19 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
// in a separate thread, and we want to prevent more reads
acquireSendInProgress();
dispatchMessagesThread.execute(() -> {
- handleSendingMessagesAndReadingMore(readType, entries, false, totalBytesSize);
+ if (sendMessagesToConsumers(readType, entries, false)) {
+ updatePendingBytesToDispatch(-size);
+ readMoreEntries();
+ } else {
+ updatePendingBytesToDispatch(-size);
+ }
});
} else {
- handleSendingMessagesAndReadingMore(readType, entries, true, totalBytesSize);
- }
- }
-
- private synchronized void handleSendingMessagesAndReadingMore(ReadType readType, List<Entry> entries,
- boolean needAcquireSendInProgress,
- long totalBytesSize) {
- boolean triggerReadingMore = sendMessagesToConsumers(readType, entries, needAcquireSendInProgress);
- int entriesProcessed = lastNumberOfEntriesProcessed;
- updatePendingBytesToDispatch(-totalBytesSize);
- boolean canReadMoreImmediately = false;
- if (entriesProcessed > 0 || skipNextBackoff) {
- // Reset the backoff when messages were processed
- retryBackoff.reset();
- // Reset the possible flag to skip the backoff delay
- skipNextBackoff = false;
- canReadMoreImmediately = true;
- }
- if (triggerReadingMore) {
- if (canReadMoreImmediately) {
- // Call readMoreEntries in the same thread to trigger the next read
- readMoreEntries();
+ if (sendMessagesToConsumers(readType, entries, true)) {
+ updatePendingBytesToDispatch(-size);
+ readMoreEntriesAsync();
} else {
- // reschedule a new read with an increasing backoff delay
- reScheduleReadWithBackoff();
+ updatePendingBytesToDispatch(-size);
}
}
}
@@ -826,7 +722,6 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
if (needTrimAckedMessages()) {
cursor.trimDeletedEntries(entries);
}
- lastNumberOfEntriesProcessed = 0;
int entriesToDispatch = entries.size();
// Trigger read more messages
@@ -837,16 +732,8 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
int remainingMessages = 0;
boolean hasChunk = false;
for (int i = 0; i < metadataArray.length; i++) {
- Entry entry = entries.get(i);
- MessageMetadata metadata;
- if (entry instanceof EntryAndMetadata) {
- metadata = ((EntryAndMetadata) entry).getMetadata();
- } else {
- metadata = Commands.peekAndCopyMessageMetadata(entry.getDataBuffer(), subscription.toString(), -1);
- // cache the metadata in the entry with EntryAndMetadata for later use to avoid re-parsing the metadata
- // and to carry the metadata and calculated stickyKeyHash with the entry
- entries.set(i, EntryAndMetadata.create(entry, metadata));
- }
+ final MessageMetadata metadata = Commands.peekAndCopyMessageMetadata(
+ entries.get(i).getDataBuffer(), subscription.toString(), -1);
if (metadata != null) {
remainingMessages += metadata.getNumMessagesInBatch();
if (!hasChunk && metadata.hasUuid()) {
@@ -863,7 +750,6 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
long totalMessagesSent = 0;
long totalBytesSent = 0;
long totalEntries = 0;
- long totalEntriesProcessed = 0;
int avgBatchSizePerMsg = remainingMessages > 0 ? Math.max(remainingMessages / entries.size(), 1) : 1;
// If the dispatcher is closed, firstAvailableConsumerPermits will be 0, which skips dispatching the
@@ -875,33 +761,27 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
log.info("[{}] rewind because no available consumer found from total {}", name, consumerList.size());
entries.subList(start, entries.size()).forEach(Entry::release);
cursor.rewind();
- lastNumberOfEntriesProcessed = (int) totalEntriesProcessed;
return false;
}
// round-robin dispatch batch size for this consumer
int availablePermits = c.isWritable() ? c.getAvailablePermits() : 1;
+ if (c.getMaxUnackedMessages() > 0) {
+ // Avoid negative number
+ int remainUnAckedMessages = Math.max(c.getMaxUnackedMessages() - c.getUnackedMessages(), 0);
+ availablePermits = Math.min(availablePermits, remainUnAckedMessages);
+ }
if (log.isDebugEnabled() && !c.isWritable()) {
log.debug("[{}-{}] consumer is not writable. dispatching only 1 message to {}; "
+ "availablePermits are {}", topic.getName(), name,
c, c.getAvailablePermits());
}
- int maxMessagesInThisBatch =
- Math.max(remainingMessages, serviceConfig.getDispatcherMaxRoundRobinBatchSize());
- if (c.getMaxUnackedMessages() > 0) {
- // Calculate the maximum number of additional unacked messages allowed
- int maxAdditionalUnackedMessages = Math.max(c.getMaxUnackedMessages() - c.getUnackedMessages(), 0);
- maxMessagesInThisBatch = Math.min(maxMessagesInThisBatch, maxAdditionalUnackedMessages);
- }
- int maxEntriesInThisBatch = Math.min(availablePermits,
- // use the average batch size per message to calculate the number of entries to
- // dispatch. round up to the next integer without using floating point arithmetic.
- (maxMessagesInThisBatch + avgBatchSizePerMsg - 1) / avgBatchSizePerMsg);
- // pick at least one entry to dispatch
- maxEntriesInThisBatch = Math.max(maxEntriesInThisBatch, 1);
-
- int end = Math.min(start + maxEntriesInThisBatch, entries.size());
+ int messagesForC = Math.min(Math.min(remainingMessages, availablePermits),
+ serviceConfig.getDispatcherMaxRoundRobinBatchSize());
+ messagesForC = Math.max(messagesForC / avgBatchSizePerMsg, 1);
+
+ int end = Math.min(start + messagesForC, entries.size());
List<Entry> entriesForThisConsumer = entries.subList(start, end);
// remove positions first from replay list first : sendMessages recycles entries
@@ -915,19 +795,17 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
EntryBatchSizes batchSizes = EntryBatchSizes.get(entriesForThisConsumer.size());
EntryBatchIndexesAcks batchIndexesAcks = EntryBatchIndexesAcks.get(entriesForThisConsumer.size());
-
totalEntries += filterEntriesForConsumer(metadataArray, start,
entriesForThisConsumer, batchSizes, sendMessageInfo, batchIndexesAcks, cursor,
readType == ReadType.Replay, c);
- totalEntriesProcessed += entriesForThisConsumer.size();
c.sendMessages(entriesForThisConsumer, batchSizes, batchIndexesAcks, sendMessageInfo.getTotalMessages(),
sendMessageInfo.getTotalBytes(), sendMessageInfo.getTotalChunkedMessages(), redeliveryTracker);
int msgSent = sendMessageInfo.getTotalMessages();
remainingMessages -= msgSent;
- start += maxEntriesInThisBatch;
- entriesToDispatch -= maxEntriesInThisBatch;
+ start += messagesForC;
+ entriesToDispatch -= messagesForC;
TOTAL_AVAILABLE_PERMITS_UPDATER.addAndGet(this,
-(msgSent - batchIndexesAcks.getTotalAckedIndexCount()));
if (log.isDebugEnabled()) {
@@ -939,7 +817,6 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
totalBytesSent += sendMessageInfo.getTotalBytes();
}
- lastNumberOfEntriesProcessed = (int) totalEntriesProcessed;
acquirePermitsForDeliveredMessages(topic, cursor, totalEntries, totalMessagesSent, totalBytesSent);
if (entriesToDispatch > 0) {
@@ -947,19 +824,15 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
log.debug("[{}] No consumers found with available permits, storing {} positions for later replay", name,
entries.size() - start);
}
- entries.subList(start, entries.size()).forEach(this::addEntryToReplay);
+ entries.subList(start, entries.size()).forEach(entry -> {
+ long stickyKeyHash = getStickyKeyHash(entry);
+ addMessageToReplay(entry.getLedgerId(), entry.getEntryId(), stickyKeyHash);
+ entry.release();
+ });
}
-
return true;
}
- protected boolean addEntryToReplay(Entry entry) {
- long stickyKeyHash = getStickyKeyHash(entry);
- boolean addedToReplay = addMessageToReplay(entry.getLedgerId(), entry.getEntryId(), stickyKeyHash);
- entry.release();
- return addedToReplay;
- }
-
private boolean sendChunkedMessagesToConsumers(ReadType readType,
List<Entry> entries,
MessageMetadata[] metadataArray) {
@@ -973,9 +846,7 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
long totalMessagesSent = 0;
long totalBytesSent = 0;
long totalEntries = 0;
- long totalEntriesProcessed = 0;
final AtomicInteger numConsumers = new AtomicInteger(assignResult.size());
- boolean notifyAddedToReplay = false;
for (Map.Entry<Consumer, List<EntryAndMetadata>> current : assignResult.entrySet()) {
final Consumer consumer = current.getKey();
final List<EntryAndMetadata> entryAndMetadataList = current.getValue();
@@ -987,7 +858,7 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
if (messagesForC < entryAndMetadataList.size()) {
for (int i = messagesForC; i < entryAndMetadataList.size(); i++) {
final EntryAndMetadata entry = entryAndMetadataList.get(i);
- notifyAddedToReplay |= addEntryToReplay(entry);
+ addMessageToReplay(entry);
entryAndMetadataList.set(i, null);
}
}
@@ -1005,13 +876,12 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
totalEntries += filterEntriesForConsumer(entryAndMetadataList, batchSizes, sendMessageInfo,
batchIndexesAcks, cursor, readType == ReadType.Replay, consumer);
- totalEntriesProcessed += entryAndMetadataList.size();
consumer.sendMessages(entryAndMetadataList, batchSizes, batchIndexesAcks,
sendMessageInfo.getTotalMessages(), sendMessageInfo.getTotalBytes(),
sendMessageInfo.getTotalChunkedMessages(), getRedeliveryTracker()
).addListener(future -> {
if (future.isDone() && numConsumers.decrementAndGet() == 0) {
- readMoreEntriesAsync();
+ readMoreEntries();
}
});
@@ -1021,11 +891,9 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
totalBytesSent += sendMessageInfo.getTotalBytes();
}
- lastNumberOfEntriesProcessed = (int) totalEntriesProcessed;
acquirePermitsForDeliveredMessages(topic, cursor, totalEntries, totalMessagesSent, totalBytesSent);
- // trigger a new readMoreEntries() call
- return numConsumers.get() == 0 || notifyAddedToReplay;
+ return numConsumers.get() == 0; // trigger a new readMoreEntries() call
}
@Override
@@ -1074,7 +942,7 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
} else {
havePendingReplayRead = false;
if (exception instanceof ManagedLedgerException.InvalidReplayPositionException) {
- Position markDeletePosition = cursor.getMarkDeletedPosition();
+ Position markDeletePosition = (Position) cursor.getMarkDeletedPosition();
redeliveryMessages.removeAllUpTo(markDeletePosition.getLedgerId(), markDeletePosition.getEntryId());
}
}
@@ -1089,7 +957,7 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
@VisibleForTesting
void scheduleReadEntriesWithDelay(Exception e, ReadType readType, long waitTimeMillis) {
topic.getBrokerService().executor().schedule(() -> {
- synchronized (PersistentDispatcherMultipleConsumers.this) {
+ synchronized (PersistentDispatcherMultipleConsumersClassic.this) {
// If it's a replay read we need to retry even if there's already
// another scheduled read, otherwise we'd be stuck until
// more messages are published.
@@ -1108,7 +976,7 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
return false;
} else {
return lastIndividualDeletedRangeFromCursorRecovery.upperEndpoint()
- .compareTo(cursor.getReadPosition()) > 0;
+ .compareTo((Position) cursor.getReadPosition()) > 0;
}
}
@@ -1128,7 +996,7 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
return 0;
}
for (Consumer consumer : consumerList) {
- if (consumer != null && !consumer.isBlocked() && consumer.cnx().isActive()) {
+ if (consumer != null && !consumer.isBlocked()) {
int availablePermits = consumer.getAvailablePermits();
if (availablePermits > 0) {
return availablePermits;
@@ -1152,45 +1020,36 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
@Override
public boolean isConsumerAvailable(Consumer consumer) {
- return consumer != null && !consumer.isBlocked() && consumer.cnx().isActive()
- && consumer.getAvailablePermits() > 0;
+ return consumer != null && !consumer.isBlocked() && consumer.getAvailablePermits() > 0;
}
@Override
public synchronized void redeliverUnacknowledgedMessages(Consumer consumer, long consumerEpoch) {
- if (log.isDebugEnabled()) {
- log.debug("[{}-{}] Redelivering unacknowledged messages for consumer {}", name, consumer,
- redeliveryMessages);
- }
- MutableBoolean addedToReplay = new MutableBoolean(false);
consumer.getPendingAcks().forEach((ledgerId, entryId, batchSize, stickyKeyHash) -> {
if (addMessageToReplay(ledgerId, entryId, stickyKeyHash)) {
redeliveryTracker.incrementAndGetRedeliveryCount((PositionFactory.create(ledgerId, entryId)));
- addedToReplay.setTrue();
}
});
- if (addedToReplay.booleanValue()) {
- notifyRedeliveryMessageAdded();
+ if (log.isDebugEnabled()) {
+ log.debug("[{}-{}] Redelivering unacknowledged messages for consumer {}", name, consumer,
+ redeliveryMessages);
}
+ readMoreEntries();
}
@Override
public synchronized void redeliverUnacknowledgedMessages(Consumer consumer, List<Position> positions) {
- if (log.isDebugEnabled()) {
- log.debug("[{}-{}] Redelivering unacknowledged messages for consumer {}", name, consumer, positions);
- }
- MutableBoolean addedToReplay = new MutableBoolean(false);
positions.forEach(position -> {
// TODO: We want to pass a sticky key hash as a third argument to guarantee the order of the messages
// on Key_Shared subscription, but it's difficult to get the sticky key here
if (addMessageToReplay(position.getLedgerId(), position.getEntryId())) {
redeliveryTracker.incrementAndGetRedeliveryCount(position);
- addedToReplay.setTrue();
}
});
- if (addedToReplay.booleanValue()) {
- notifyRedeliveryMessageAdded();
+ if (log.isDebugEnabled()) {
+ log.debug("[{}-{}] Redelivering unacknowledged messages for consumer {}", name, consumer, positions);
}
+ readMoreEntries();
}
@Override
@@ -1351,20 +1210,16 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
delayedDeliveryTracker.get().getScheduledMessages(maxMessagesToRead);
messagesAvailableNow.forEach(p -> redeliveryMessages.add(p.getLedgerId(), p.getEntryId()));
}
+
if (!redeliveryMessages.isEmpty()) {
- return redeliveryMessages.getMessagesToReplayNow(maxMessagesToRead, createFilterForReplay());
+ return redeliveryMessages.getMessagesToReplayNow(maxMessagesToRead, position -> true);
} else {
return Collections.emptyNavigableSet();
}
}
- protected Optional<Position> getFirstPositionInReplay() {
- return redeliveryMessages.getFirstPositionInReplay();
- }
-
/**
- * Creates a stateful filter for filtering replay positions.
- * This is only used for Key_Shared mode to skip replaying certain entries.
+ * This is a mode method designed for Key_Shared mode.
* Filter out the entries that will be discarded due to the order guarantee mechanism of Key_Shared mode.
* This method is in order to avoid the scenario below:
* - Get positions from the Replay queue.
@@ -1372,20 +1227,18 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
* - The order guarantee mechanism of Key_Shared mode filtered out all the entries.
* - Delivery non entry to the client, but we did a BK read.
*/
- protected Predicate<Position> createFilterForReplay() {
- // pick all positions from the replay
- return position -> true;
+ protected NavigableSet<Position> filterOutEntriesWillBeDiscarded(NavigableSet<Position> src) {
+ return src;
}
/**
- * Checks if the dispatcher is allowed to read messages from the cursor.
+ * This is a mode method designed for Key_Shared mode, to avoid unnecessary stuck.
+ * See detail {@link PersistentStickyKeyDispatcherMultipleConsumersClassic#hasConsumersNeededNormalRead}.
*/
- protected boolean isNormalReadAllowed() {
+ protected boolean hasConsumersNeededNormalRead() {
return true;
}
-
-
protected synchronized boolean shouldPauseDeliveryForDelayTracker() {
return delayedDeliveryTracker.isPresent() && delayedDeliveryTracker.get().shouldPauseAllDeliveries();
}
@@ -1421,6 +1274,11 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
}
}
+ private void addMessageToReplay(Entry entry) {
+ addMessageToReplay(entry.getLedgerId(), entry.getEntryId());
+ entry.release();
+ }
+
protected boolean addMessageToReplay(long ledgerId, long entryId, long stickyKeyHash) {
if (checkIfMessageIsUnacked(ledgerId, entryId)) {
redeliveryMessages.add(ledgerId, entryId, stickyKeyHash);
@@ -1430,13 +1288,6 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
}
}
- /**
- * Notify the dispatcher that a message has been added to the redelivery list.
- */
- private void notifyRedeliveryMessageAdded() {
- readMoreEntriesAsync();
- }
-
protected boolean addMessageToReplay(long ledgerId, long entryId) {
if (checkIfMessageIsUnacked(ledgerId, entryId)) {
redeliveryMessages.add(ledgerId, entryId);
@@ -1461,7 +1312,7 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
if (totalAvailablePermits > 0 && !havePendingReplayRead && !havePendingRead
&& cursor.getNumberOfEntriesInBacklog(false) > 0) {
log.warn("{}-{} Dispatcher is stuck and unblocking by issuing reads", topic.getName(), name);
- readMoreEntriesAsync();
+ readMoreEntries();
return true;
}
return false;
@@ -1488,13 +1339,17 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
return Collections.emptyMap();
}
+ @Override
+ public boolean isClassic() {
+ return true;
+ }
+
public ManagedCursor getCursor() {
return cursor;
}
protected int getStickyKeyHash(Entry entry) {
- // There's no need to calculate the hash for Shared subscription
- return STICKY_KEY_HASH_NOT_SET;
+ return StickyKeyConsumerSelector.STICKY_KEY_HASH_NOT_SET;
}
public Subscription getSubscription() {
@@ -1505,5 +1360,15 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
return redeliveryMessages.size();
}
- private static final Logger log = LoggerFactory.getLogger(PersistentDispatcherMultipleConsumers.class);
+ @Override
+ public boolean isHavePendingRead() {
+ return havePendingRead;
+ }
+
+ @Override
+ public boolean isHavePendingReplayRead() {
+ return havePendingReplayRead;
+ }
+
+ private static final Logger log = LoggerFactory.getLogger(PersistentDispatcherMultipleConsumersClassic.class);
}
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 925e99ed699..df053e6d8a5 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
@@ -50,6 +50,7 @@ import org.apache.pulsar.broker.service.ImpactedConsumersResult;
import org.apache.pulsar.broker.service.PendingAcksMap;
import org.apache.pulsar.broker.service.SendMessageInfo;
import org.apache.pulsar.broker.service.StickyKeyConsumerSelector;
+import org.apache.pulsar.broker.service.StickyKeyDispatcher;
import org.apache.pulsar.broker.service.Subscription;
import org.apache.pulsar.client.api.Range;
import org.apache.pulsar.common.api.proto.CommandSubscribe.SubType;
@@ -60,7 +61,8 @@ import org.apache.pulsar.common.util.FutureUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public class PersistentStickyKeyDispatcherMultipleConsumers extends PersistentDispatcherMultipleConsumers {
+public class PersistentStickyKeyDispatcherMultipleConsumers extends PersistentDispatcherMultipleConsumers implements
+ StickyKeyDispatcher {
private final boolean allowOutOfOrderDelivery;
private final StickyKeyConsumerSelector selector;
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
new file mode 100644
index 00000000000..c227bf5b435
--- /dev/null
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersClassic.java
@@ -0,0 +1,583 @@
+/*
+ * 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 com.google.common.annotations.VisibleForTesting;
+import io.netty.util.concurrent.FastThreadLocal;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.NavigableSet;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+import org.apache.bookkeeper.mledger.Entry;
+import org.apache.bookkeeper.mledger.ManagedCursor;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
+import org.apache.commons.collections4.MapUtils;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.service.BrokerServiceException;
+import org.apache.pulsar.broker.service.ConsistentHashingStickyKeyConsumerSelector;
+import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.EntryAndMetadata;
+import org.apache.pulsar.broker.service.EntryBatchIndexesAcks;
+import org.apache.pulsar.broker.service.EntryBatchSizes;
+import org.apache.pulsar.broker.service.HashRangeAutoSplitStickyKeyConsumerSelector;
+import org.apache.pulsar.broker.service.HashRangeExclusiveStickyKeyConsumerSelector;
+import org.apache.pulsar.broker.service.SendMessageInfo;
+import org.apache.pulsar.broker.service.StickyKeyConsumerSelector;
+import org.apache.pulsar.broker.service.StickyKeyDispatcher;
+import org.apache.pulsar.broker.service.Subscription;
+import org.apache.pulsar.client.api.Range;
+import org.apache.pulsar.common.api.proto.CommandSubscribe.SubType;
+import org.apache.pulsar.common.api.proto.KeySharedMeta;
+import org.apache.pulsar.common.api.proto.KeySharedMode;
+import org.apache.pulsar.common.util.FutureUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This is the "classic" dispatcher implementation for the Key_Shared subscription that was used before
+ * Pulsar 4.0.0 and PIP-379. When `subscriptionKeySharedUseClassicPersistentImplementation=true`,
+ * this dispatcher will be used. The main purpose is to provide a way to rollback to the old behavior
+ * in case of issues with the preferred implementation.
+ */
+public class PersistentStickyKeyDispatcherMultipleConsumersClassic
+ extends PersistentDispatcherMultipleConsumersClassic implements StickyKeyDispatcher {
+
+ private final boolean allowOutOfOrderDelivery;
+ private final StickyKeyConsumerSelector selector;
+
+ private boolean isDispatcherStuckOnReplays = false;
+ private final KeySharedMode keySharedMode;
+
+ /**
+ * When a consumer joins, it will be added to this map with the current read position.
+ * This means that, in order to preserve ordering, new consumers can only receive old
+ * messages, until the mark-delete position will move past this point.
+ */
+ private final LinkedHashMap<Consumer, Position> recentlyJoinedConsumers;
+
+ PersistentStickyKeyDispatcherMultipleConsumersClassic(PersistentTopic topic, ManagedCursor cursor,
+ Subscription subscription, ServiceConfiguration conf,
+ KeySharedMeta ksm) {
+ super(topic, cursor, subscription, ksm.isAllowOutOfOrderDelivery());
+
+ this.allowOutOfOrderDelivery = ksm.isAllowOutOfOrderDelivery();
+ this.recentlyJoinedConsumers = allowOutOfOrderDelivery ? null : new LinkedHashMap<>();
+ this.keySharedMode = ksm.getKeySharedMode();
+ switch (this.keySharedMode) {
+ case AUTO_SPLIT:
+ if (conf.isSubscriptionKeySharedUseConsistentHashing()) {
+ selector = new ConsistentHashingStickyKeyConsumerSelector(
+ conf.getSubscriptionKeySharedConsistentHashingReplicaPoints(),
+ false,
+ // Classic implementation uses Integer.MAX_VALUE - 1 as the range end value
+ Integer.MAX_VALUE - 1);
+ } else {
+ selector = new HashRangeAutoSplitStickyKeyConsumerSelector();
+ }
+ break;
+
+ case STICKY:
+ this.selector = new HashRangeExclusiveStickyKeyConsumerSelector();
+ break;
+
+ default:
+ throw new IllegalArgumentException("Invalid key-shared mode: " + keySharedMode);
+ }
+ }
+
+ @VisibleForTesting
+ public StickyKeyConsumerSelector getSelector() {
+ return selector;
+ }
+
+ @Override
+ public synchronized CompletableFuture<Void> addConsumer(Consumer consumer) {
+ if (IS_CLOSED_UPDATER.get(this) == TRUE) {
+ log.warn("[{}] Dispatcher is already closed. Closing consumer {}", name, consumer);
+ consumer.disconnect();
+ return CompletableFuture.completedFuture(null);
+ }
+ return super.addConsumer(consumer).thenCompose(__ ->
+ selector.addConsumer(consumer).handle((result, ex) -> {
+ if (ex != null) {
+ synchronized (PersistentStickyKeyDispatcherMultipleConsumersClassic.this) {
+ consumerSet.removeAll(consumer);
+ consumerList.remove(consumer);
+ }
+ throw FutureUtil.wrapToCompletionException(ex);
+ }
+ return result;
+ })
+ ).thenRun(() -> {
+ synchronized (PersistentStickyKeyDispatcherMultipleConsumersClassic.this) {
+ Position readPositionWhenJoining = (Position) cursor.getReadPosition();
+ consumer.setReadPositionWhenJoining(readPositionWhenJoining);
+ // If this was the 1st consumer, or if all the messages are already acked, then we
+ // don't need to do anything special
+ if (!allowOutOfOrderDelivery
+ && recentlyJoinedConsumers != null
+ && consumerList.size() > 1
+ && cursor.getNumberOfEntriesSinceFirstNotAckedMessage() > 1) {
+ recentlyJoinedConsumers.put(consumer, readPositionWhenJoining);
+ }
+ }
+ });
+ }
+
+ @Override
+ public synchronized void removeConsumer(Consumer consumer) throws BrokerServiceException {
+ // 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,
+ // the broker will try to redeliver the messages to the consumer that has already been closed.
+ // As a result, the messages are not redelivered to any consumer, and the mark-delete position does not move,
+ // eventually causing all consumers to get stuck.
+ selector.removeConsumer(consumer);
+ super.removeConsumer(consumer);
+ if (recentlyJoinedConsumers != null) {
+ recentlyJoinedConsumers.remove(consumer);
+ if (consumerList.size() == 1) {
+ recentlyJoinedConsumers.clear();
+ }
+ if (removeConsumersFromRecentJoinedConsumers() || !redeliveryMessages.isEmpty()) {
+ readMoreEntries();
+ }
+ }
+ }
+
+ private static final FastThreadLocal<Map<Consumer, List<Entry>>> localGroupedEntries =
+ new FastThreadLocal<Map<Consumer, List<Entry>>>() {
+ @Override
+ protected Map<Consumer, List<Entry>> initialValue() throws Exception {
+ return new HashMap<>();
+ }
+ };
+
+ private static final FastThreadLocal<Map<Consumer, List<Position>>> localGroupedPositions =
+ new FastThreadLocal<Map<Consumer, List<Position>>>() {
+ @Override
+ protected Map<Consumer, List<Position>> initialValue() throws Exception {
+ return new HashMap<>();
+ }
+ };
+
+ @Override
+ protected synchronized boolean trySendMessagesToConsumers(ReadType readType, List<Entry> entries) {
+ long totalMessagesSent = 0;
+ long totalBytesSent = 0;
+ long totalEntries = 0;
+ int entriesCount = entries.size();
+
+ // Trigger read more messages
+ if (entriesCount == 0) {
+ return true;
+ }
+
+ if (consumerSet.isEmpty()) {
+ entries.forEach(Entry::release);
+ cursor.rewind();
+ return false;
+ }
+
+ // A corner case that we have to retry a readMoreEntries in order to preserver order delivery.
+ // This may happen when consumer closed. See issue #12885 for details.
+ if (!allowOutOfOrderDelivery) {
+ NavigableSet<Position> messagesToReplayNow = this.getMessagesToReplayNow(1);
+ if (messagesToReplayNow != null && !messagesToReplayNow.isEmpty()) {
+ Position replayPosition = messagesToReplayNow.first();
+
+ // We have received a message potentially from the delayed tracker and, since we're not using it
+ // right now, it needs to be added to the redelivery tracker or we won't attempt anymore to
+ // resend it (until we disconnect consumer).
+ redeliveryMessages.add(replayPosition.getLedgerId(), replayPosition.getEntryId());
+
+ if (this.minReplayedPosition != null) {
+ // If relayPosition is a new entry wither smaller position is inserted for redelivery during this
+ // async read, it is possible that this relayPosition should dispatch to consumer first. So in
+ // order to preserver order delivery, we need to discard this read result, and try to trigger a
+ // replay read, that containing "relayPosition", by calling readMoreEntries.
+ if (replayPosition.compareTo(minReplayedPosition) < 0) {
+ if (log.isDebugEnabled()) {
+ log.debug("[{}] Position {} (<{}) is inserted for relay during current {} read, "
+ + "discard this read and retry with readMoreEntries.",
+ name, replayPosition, minReplayedPosition, readType);
+ }
+ if (readType == ReadType.Normal) {
+ entries.forEach(entry -> {
+ long stickyKeyHash = getStickyKeyHash(entry);
+ addMessageToReplay(entry.getLedgerId(), entry.getEntryId(), stickyKeyHash);
+ entry.release();
+ });
+ } else if (readType == ReadType.Replay) {
+ entries.forEach(Entry::release);
+ }
+ return true;
+ }
+ }
+ }
+ }
+
+ final Map<Consumer, List<Entry>> groupedEntries = localGroupedEntries.get();
+ groupedEntries.clear();
+ final Map<Consumer, Set<Integer>> consumerStickyKeyHashesMap = new HashMap<>();
+
+ for (Entry entry : entries) {
+ int stickyKeyHash = getStickyKeyHash(entry);
+ Consumer c = selector.select(stickyKeyHash);
+ if (c != null) {
+ groupedEntries.computeIfAbsent(c, k -> new ArrayList<>()).add(entry);
+ consumerStickyKeyHashesMap.computeIfAbsent(c, k -> new HashSet<>()).add(stickyKeyHash);
+ } else {
+ addMessageToReplay(entry.getLedgerId(), entry.getEntryId(), stickyKeyHash);
+ entry.release();
+ }
+ }
+
+ AtomicInteger keyNumbers = new AtomicInteger(groupedEntries.size());
+
+ int currentThreadKeyNumber = groupedEntries.size();
+ if (currentThreadKeyNumber == 0) {
+ currentThreadKeyNumber = -1;
+ }
+ for (Map.Entry<Consumer, List<Entry>> current : groupedEntries.entrySet()) {
+ Consumer consumer = current.getKey();
+ assert consumer != null; // checked when added to groupedEntries
+ List<Entry> entriesWithSameKey = current.getValue();
+ int entriesWithSameKeyCount = entriesWithSameKey.size();
+ int availablePermits = getAvailablePermits(consumer);
+ int messagesForC = getRestrictedMaxEntriesForConsumer(consumer,
+ entriesWithSameKey.stream().map(Entry::getPosition).collect(Collectors.toList()), availablePermits,
+ readType, consumerStickyKeyHashesMap.get(consumer));
+ if (log.isDebugEnabled()) {
+ log.debug("[{}] select consumer {} with messages num {}, read type is {}",
+ name, consumer.consumerName(), messagesForC, readType);
+ }
+
+ if (messagesForC < entriesWithSameKeyCount) {
+ // We are not able to push all the messages with given key to its consumer,
+ // so we discard for now and mark them for later redelivery
+ for (int i = messagesForC; i < entriesWithSameKeyCount; i++) {
+ Entry entry = entriesWithSameKey.get(i);
+ long stickyKeyHash = getStickyKeyHash(entry);
+ addMessageToReplay(entry.getLedgerId(), entry.getEntryId(), stickyKeyHash);
+ entry.release();
+ entriesWithSameKey.set(i, null);
+ }
+ }
+
+ if (messagesForC > 0) {
+ // remove positions first from replay list first : sendMessages recycles entries
+ if (readType == ReadType.Replay) {
+ for (int i = 0; i < messagesForC; i++) {
+ Entry entry = entriesWithSameKey.get(i);
+ redeliveryMessages.remove(entry.getLedgerId(), entry.getEntryId());
+ }
+ }
+
+ SendMessageInfo sendMessageInfo = SendMessageInfo.getThreadLocal();
+ EntryBatchSizes batchSizes = EntryBatchSizes.get(messagesForC);
+ EntryBatchIndexesAcks batchIndexesAcks = EntryBatchIndexesAcks.get(messagesForC);
+ totalEntries += filterEntriesForConsumer(entriesWithSameKey, batchSizes, sendMessageInfo,
+ batchIndexesAcks, cursor, readType == ReadType.Replay, consumer);
+ consumer.sendMessages(entriesWithSameKey, batchSizes, batchIndexesAcks,
+ sendMessageInfo.getTotalMessages(),
+ sendMessageInfo.getTotalBytes(), sendMessageInfo.getTotalChunkedMessages(),
+ getRedeliveryTracker()).addListener(future -> {
+ if (future.isDone() && keyNumbers.decrementAndGet() == 0) {
+ readMoreEntries();
+ }
+ });
+
+ TOTAL_AVAILABLE_PERMITS_UPDATER.getAndAdd(this,
+ -(sendMessageInfo.getTotalMessages() - batchIndexesAcks.getTotalAckedIndexCount()));
+ totalMessagesSent += sendMessageInfo.getTotalMessages();
+ totalBytesSent += sendMessageInfo.getTotalBytes();
+ } else {
+ currentThreadKeyNumber = keyNumbers.decrementAndGet();
+ }
+ }
+
+ // acquire message-dispatch permits for already delivered messages
+ acquirePermitsForDeliveredMessages(topic, cursor, totalEntries, totalMessagesSent, totalBytesSent);
+
+ if (totalMessagesSent == 0 && (recentlyJoinedConsumers == null || recentlyJoinedConsumers.isEmpty())) {
+ // This means, that all the messages we've just read cannot be dispatched right now.
+ // This condition can only happen when:
+ // 1. We have consumers ready to accept messages (otherwise the would not haven been triggered)
+ // 2. All keys in the current set of messages are routing to consumers that are currently busy
+ //
+ // The solution here is to move on and read next batch of messages which might hopefully contain
+ // also keys meant for other consumers.
+ //
+ // We do it unless that are "recently joined consumers". In that case, we would be looking
+ // ahead in the stream while the new consumers are not ready to accept the new messages,
+ // therefore would be most likely only increase the distance between read-position and mark-delete
+ // position.
+ isDispatcherStuckOnReplays = true;
+ return true;
+ } else if (currentThreadKeyNumber == 0) {
+ return true;
+ }
+ return false;
+ }
+
+ private int getRestrictedMaxEntriesForConsumer(Consumer consumer, List<? extends Position> entries,
+ int availablePermits, ReadType readType, Set<Integer> stickyKeyHashes) {
+ int maxMessages = Math.min(entries.size(), availablePermits);
+ if (maxMessages == 0) {
+ return 0;
+ }
+ if (readType == ReadType.Normal && stickyKeyHashes != null
+ && redeliveryMessages.containsStickyKeyHashes(stickyKeyHashes)) {
+ // If redeliveryMessages contains messages that correspond to the same hash as the messages
+ // that the dispatcher is trying to send, do not send those messages for order guarantee
+ return 0;
+ }
+ if (recentlyJoinedConsumers == null) {
+ return maxMessages;
+ }
+ removeConsumersFromRecentJoinedConsumers();
+ Position maxReadPosition = recentlyJoinedConsumers.get(consumer);
+ // At this point, all the old messages were already consumed and this consumer
+ // is now ready to receive any message
+ if (maxReadPosition == null) {
+ // The consumer has not recently joined, so we can send all messages
+ return maxMessages;
+ }
+
+ // If the read type is Replay, we should avoid send messages that hold by other consumer to the new consumers,
+ // For example, we have 10 messages [0,1,2,3,4,5,6,7,8,9]
+ // If the consumer0 get message 0 and 1, and does not acked message 0, then consumer1 joined,
+ // when consumer1 get message 2,3, the broker will not dispatch messages to consumer1
+ // because of the mark delete position did not move forward.
+ // So message 2,3 will stored in the redeliver tracker.
+ // Now, consumer2 joined, it will read new messages from the cursor,
+ // so the recentJoinedPosition is 4 for consumer2
+ // Because of there are messages need to redeliver, so the broker will read the redelivery message first [2,3]
+ // message [2,3] is lower than the recentJoinedPosition 4,
+ // so the message [2,3] will dispatched to the consumer2
+ // But the message [2,3] should not dispatch to consumer2.
+
+ if (readType == ReadType.Replay) {
+ Position minReadPositionForRecentJoinedConsumer = recentlyJoinedConsumers.values().iterator().next();
+ if (minReadPositionForRecentJoinedConsumer != null
+ && minReadPositionForRecentJoinedConsumer.compareTo(maxReadPosition) < 0) {
+ maxReadPosition = minReadPositionForRecentJoinedConsumer;
+ }
+ }
+ // Here, the consumer is one that has recently joined, so we can only send messages that were
+ // published before it has joined.
+ for (int i = 0; i < maxMessages; i++) {
+ if (((Position) entries.get(i)).compareTo(maxReadPosition) >= 0) {
+ // We have already crossed the divider line. All messages in the list are now
+ // newer than what we can currently dispatch to this consumer
+ return i;
+ }
+ }
+
+ return maxMessages;
+ }
+
+ @Override
+ public void markDeletePositionMoveForward() {
+ // Execute the notification in different thread to avoid a mutex chain here
+ // from the delete operation that was completed
+ topic.getBrokerService().getTopicOrderedExecutor().execute(() -> {
+ synchronized (PersistentStickyKeyDispatcherMultipleConsumersClassic.this) {
+ if (recentlyJoinedConsumers != null && !recentlyJoinedConsumers.isEmpty()
+ && removeConsumersFromRecentJoinedConsumers()) {
+ // After we process acks, we need to check whether the mark-delete position was advanced and we
+ // can finally read more messages. It's safe to call readMoreEntries() multiple times.
+ readMoreEntries();
+ }
+ }
+ });
+ }
+
+ private boolean removeConsumersFromRecentJoinedConsumers() {
+ if (MapUtils.isEmpty(recentlyJoinedConsumers)) {
+ return false;
+ }
+ Iterator<Map.Entry<Consumer, Position>> itr = recentlyJoinedConsumers.entrySet().iterator();
+ boolean hasConsumerRemovedFromTheRecentJoinedConsumers = false;
+ Position mdp = (Position) cursor.getMarkDeletedPosition();
+ if (mdp != null) {
+ Position nextPositionOfTheMarkDeletePosition =
+ ((ManagedLedgerImpl) cursor.getManagedLedger()).getNextValidPosition(mdp);
+ while (itr.hasNext()) {
+ Map.Entry<Consumer, Position> entry = itr.next();
+ if (entry.getValue().compareTo(nextPositionOfTheMarkDeletePosition) <= 0) {
+ itr.remove();
+ hasConsumerRemovedFromTheRecentJoinedConsumers = true;
+ } else {
+ break;
+ }
+ }
+ }
+ return hasConsumerRemovedFromTheRecentJoinedConsumers;
+ }
+
+ @Override
+ protected synchronized NavigableSet<Position> getMessagesToReplayNow(int maxMessagesToRead) {
+ if (isDispatcherStuckOnReplays) {
+ // If we're stuck on replay, we want to move forward reading on the topic (until the overall max-unacked
+ // messages kicks in), instead of keep replaying the same old messages, since the consumer that these
+ // messages are routing to might be busy at the moment
+ this.isDispatcherStuckOnReplays = false;
+ return Collections.emptyNavigableSet();
+ } else {
+ return super.getMessagesToReplayNow(maxMessagesToRead);
+ }
+ }
+
+ private int getAvailablePermits(Consumer c) {
+ int availablePermits = Math.max(c.getAvailablePermits(), 0);
+ if (c.getMaxUnackedMessages() > 0) {
+ // Avoid negative number
+ int remainUnAckedMessages = Math.max(c.getMaxUnackedMessages() - c.getUnackedMessages(), 0);
+ availablePermits = Math.min(availablePermits, remainUnAckedMessages);
+ }
+ return availablePermits;
+ }
+
+ @Override
+ protected synchronized NavigableSet<Position> filterOutEntriesWillBeDiscarded(NavigableSet<Position> src) {
+ // The variable "hashesToBeBlocked" and "recentlyJoinedConsumers" will be null if "isAllowOutOfOrderDelivery()",
+ // So skip this filter out.
+ if (isAllowOutOfOrderDelivery()) {
+ return src;
+ }
+ if (src.isEmpty()) {
+ return src;
+ }
+ NavigableSet<Position> res = new TreeSet<>();
+ // Group positions.
+ final Map<Consumer, List<Position>> groupedPositions = localGroupedPositions.get();
+ groupedPositions.clear();
+ for (Position pos : src) {
+ Long stickyKeyHash = redeliveryMessages.getHash(pos.getLedgerId(), pos.getEntryId());
+ if (stickyKeyHash == null) {
+ res.add(pos);
+ continue;
+ }
+ Consumer c = selector.select(stickyKeyHash.intValue());
+ if (c == null) {
+ // Maybe using HashRangeExclusiveStickyKeyConsumerSelector.
+ continue;
+ }
+ groupedPositions.computeIfAbsent(c, k -> new ArrayList<>()).add(pos);
+ }
+ // Filter positions by the Recently Joined Position rule.
+ for (Map.Entry<Consumer, List<Position>> item : groupedPositions.entrySet()) {
+ int availablePermits = getAvailablePermits(item.getKey());
+ if (availablePermits == 0) {
+ continue;
+ }
+ int posCountToRead = getRestrictedMaxEntriesForConsumer(item.getKey(), item.getValue(), availablePermits,
+ ReadType.Replay, null);
+ if (posCountToRead > 0) {
+ res.addAll(item.getValue().subList(0, posCountToRead));
+ }
+ }
+ return res;
+ }
+
+ /**
+ * In Key_Shared mode, the consumer will not receive any entries from a normal reading if it is included in
+ * {@link #recentlyJoinedConsumers}, they can only receive entries from replay reads.
+ * If all entries in {@link #redeliveryMessages} have been filtered out due to the order guarantee mechanism,
+ * Broker need a normal read to make the consumers not included in @link #recentlyJoinedConsumers} will not be
+ * stuck. See https://github.com/apache/pulsar/pull/7105.
+ */
+ @Override
+ protected boolean hasConsumersNeededNormalRead() {
+ // The variable "hashesToBeBlocked" and "recentlyJoinedConsumers" will be null if "isAllowOutOfOrderDelivery()",
+ // So the method "filterOutEntriesWillBeDiscarded" will filter out nothing, just return "true" here.
+ if (isAllowOutOfOrderDelivery()) {
+ return true;
+ }
+ for (Consumer consumer : consumerList) {
+ if (consumer == null || consumer.isBlocked()) {
+ continue;
+ }
+ if (recentlyJoinedConsumers.containsKey(consumer)) {
+ continue;
+ }
+ if (consumer.getAvailablePermits() > 0) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ @Override
+ public SubType getType() {
+ return SubType.Key_Shared;
+ }
+
+ @Override
+ protected Set<? extends Position> asyncReplayEntries(Set<? extends Position> positions) {
+ return cursor.asyncReplayEntries(positions, this, ReadType.Replay, true);
+ }
+
+ public KeySharedMode getKeySharedMode() {
+ return this.keySharedMode;
+ }
+
+ public boolean isAllowOutOfOrderDelivery() {
+ return this.allowOutOfOrderDelivery;
+ }
+
+ public boolean hasSameKeySharedPolicy(KeySharedMeta ksm) {
+ return (ksm.getKeySharedMode() == this.keySharedMode
+ && ksm.isAllowOutOfOrderDelivery() == this.allowOutOfOrderDelivery);
+ }
+
+ public LinkedHashMap<Consumer, Position> getRecentlyJoinedConsumers() {
+ return recentlyJoinedConsumers;
+ }
+
+ public Map<Consumer, List<Range>> getConsumerKeyHashRanges() {
+ return selector.getConsumerKeyHashRanges();
+ }
+
+ @Override
+ protected int getStickyKeyHash(Entry entry) {
+ if (entry instanceof EntryAndMetadata entryAndMetadata) {
+ // use the cached sticky key hash if available, otherwise calculate the sticky key hash and cache it
+ return entryAndMetadata.getOrUpdateCachedStickyKeyHash(selector::makeStickyKeyHash);
+ }
+ return selector.makeStickyKeyHash(peekStickyKey(entry.getDataBuffer()));
+ }
+
+ private static final Logger log =
+ LoggerFactory.getLogger(PersistentStickyKeyDispatcherMultipleConsumersClassic.class);
+
+}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
index b8d351bddf8..eaa147b81b1 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
@@ -25,6 +25,7 @@ import com.google.common.base.MoreObjects;
import io.netty.buffer.ByteBuf;
import java.io.IOException;
import java.util.Collections;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -69,6 +70,7 @@ import org.apache.pulsar.broker.service.Consumer;
import org.apache.pulsar.broker.service.Dispatcher;
import org.apache.pulsar.broker.service.EntryFilterSupport;
import org.apache.pulsar.broker.service.GetStatsOptions;
+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.plugin.EntryFilter;
@@ -250,7 +252,12 @@ public class PersistentSubscription extends AbstractSubscription {
case Shared:
if (dispatcher == null || dispatcher.getType() != SubType.Shared) {
previousDispatcher = dispatcher;
- dispatcher = new PersistentDispatcherMultipleConsumers(topic, cursor, this);
+ ServiceConfiguration config = topic.getBrokerService().getPulsar().getConfig();
+ if (config.isSubscriptionSharedUseClassicPersistentImplementation()) {
+ dispatcher = new PersistentDispatcherMultipleConsumersClassic(topic, cursor, this);
+ } else {
+ dispatcher = new PersistentDispatcherMultipleConsumers(topic, cursor, this);
+ }
}
break;
case Failover:
@@ -271,11 +278,19 @@ public class PersistentSubscription extends AbstractSubscription {
case Key_Shared:
KeySharedMeta ksm = consumer.getKeySharedMeta();
if (dispatcher == null || dispatcher.getType() != SubType.Key_Shared
- || !((PersistentStickyKeyDispatcherMultipleConsumers) dispatcher)
+ || !((StickyKeyDispatcher) dispatcher)
.hasSameKeySharedPolicy(ksm)) {
previousDispatcher = dispatcher;
- dispatcher = new PersistentStickyKeyDispatcherMultipleConsumers(topic, cursor, this,
- topic.getBrokerService().getPulsar().getConfiguration(), ksm);
+ ServiceConfiguration config = topic.getBrokerService().getPulsar().getConfig();
+ if (config.isSubscriptionKeySharedUseClassicPersistentImplementation()) {
+ dispatcher =
+ new PersistentStickyKeyDispatcherMultipleConsumersClassic(topic, cursor,
+ this,
+ topic.getBrokerService().getPulsar().getConfiguration(), ksm);
+ } else {
+ dispatcher = new PersistentStickyKeyDispatcherMultipleConsumers(topic, cursor, this,
+ topic.getBrokerService().getPulsar().getConfiguration(), ksm);
+ }
}
break;
default:
@@ -1221,7 +1236,7 @@ public class PersistentSubscription extends AbstractSubscription {
Dispatcher dispatcher = this.dispatcher;
if (dispatcher != null) {
Map<Consumer, List<Range>> consumerKeyHashRanges = getType() == SubType.Key_Shared
- ? ((PersistentStickyKeyDispatcherMultipleConsumers) dispatcher).getConsumerKeyHashRanges() : null;
+ ? ((StickyKeyDispatcher) dispatcher).getConsumerKeyHashRanges() : null;
dispatcher.getConsumers().forEach(consumer -> {
ConsumerStatsImpl consumerStats = consumer.getStats();
if (!getStatsOptions.isExcludeConsumers()) {
@@ -1260,17 +1275,18 @@ public class PersistentSubscription extends AbstractSubscription {
}
}
- if (dispatcher instanceof PersistentDispatcherMultipleConsumers) {
+ if (dispatcher instanceof AbstractPersistentDispatcherMultipleConsumers) {
subStats.delayedMessageIndexSizeInBytes =
- ((PersistentDispatcherMultipleConsumers) dispatcher).getDelayedTrackerMemoryUsage();
+ ((AbstractPersistentDispatcherMultipleConsumers) dispatcher).getDelayedTrackerMemoryUsage();
subStats.bucketDelayedIndexStats =
- ((PersistentDispatcherMultipleConsumers) dispatcher).getBucketDelayedIndexStats();
+ ((AbstractPersistentDispatcherMultipleConsumers) dispatcher).getBucketDelayedIndexStats();
}
if (Subscription.isIndividualAckMode(subType)) {
- if (dispatcher instanceof PersistentDispatcherMultipleConsumers) {
- PersistentDispatcherMultipleConsumers d = (PersistentDispatcherMultipleConsumers) dispatcher;
+ if (dispatcher instanceof AbstractPersistentDispatcherMultipleConsumers) {
+ AbstractPersistentDispatcherMultipleConsumers d =
+ (AbstractPersistentDispatcherMultipleConsumers) dispatcher;
subStats.unackedMessages = d.getTotalUnackedMessages();
subStats.blockedSubscriptionOnUnackedMsgs = d.isBlockedDispatcherOnUnackedMsgs();
subStats.msgDelayed = d.getNumberOfDelayedMessages();
@@ -1290,12 +1306,18 @@ public class PersistentSubscription extends AbstractSubscription {
subStats.isReplicated = isReplicated();
subStats.subscriptionProperties = subscriptionProperties;
subStats.isDurable = cursor.isDurable();
- if (getType() == SubType.Key_Shared && dispatcher instanceof PersistentStickyKeyDispatcherMultipleConsumers) {
- PersistentStickyKeyDispatcherMultipleConsumers keySharedDispatcher =
- (PersistentStickyKeyDispatcherMultipleConsumers) dispatcher;
-
+ if (getType() == SubType.Key_Shared && dispatcher instanceof StickyKeyDispatcher) {
+ StickyKeyDispatcher keySharedDispatcher = (StickyKeyDispatcher) dispatcher;
subStats.allowOutOfOrderDelivery = keySharedDispatcher.isAllowOutOfOrderDelivery();
subStats.keySharedMode = keySharedDispatcher.getKeySharedMode().toString();
+
+ LinkedHashMap<Consumer, Position> recentlyJoinedConsumers = keySharedDispatcher
+ .getRecentlyJoinedConsumers();
+ if (recentlyJoinedConsumers != null && recentlyJoinedConsumers.size() > 0) {
+ recentlyJoinedConsumers.forEach((k, v) -> {
+ subStats.consumersAfterMarkDeletePosition.put(k.consumerName(), v.toString());
+ });
+ }
}
subStats.nonContiguousDeletedMessagesRanges = cursor.getTotalNonContiguousDeletedMessagesRange();
subStats.nonContiguousDeletedMessagesRangesSerializedSize =
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
index 3cce175660e..9c86a99de0f 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
@@ -2449,9 +2449,9 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
}
if (Subscription.isIndividualAckMode(subscription.getType())) {
- if (subscription.getDispatcher() instanceof PersistentDispatcherMultipleConsumers) {
- PersistentDispatcherMultipleConsumers dispatcher =
- (PersistentDispatcherMultipleConsumers) subscription.getDispatcher();
+ if (subscription.getDispatcher() instanceof AbstractPersistentDispatcherMultipleConsumers) {
+ AbstractPersistentDispatcherMultipleConsumers dispatcher =
+ (AbstractPersistentDispatcherMultipleConsumers) subscription.getDispatcher();
topicStatsStream.writePair("blockedSubscriptionOnUnackedMsgs",
dispatcher.isBlockedDispatcherOnUnackedMsgs());
topicStatsStream.writePair("unackedMessages",
@@ -2758,11 +2758,11 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
// subscription metrics
PersistentSubscription sub = subscriptions.get(Codec.decode(c.getName()));
if (sub != null) {
- if (sub.getDispatcher() instanceof PersistentDispatcherMultipleConsumers) {
- PersistentDispatcherMultipleConsumers dispatcher =
- (PersistentDispatcherMultipleConsumers) sub.getDispatcher();
- cs.subscriptionHavePendingRead = dispatcher.havePendingRead;
- cs.subscriptionHavePendingReplayRead = dispatcher.havePendingReplayRead;
+ if (sub.getDispatcher() instanceof AbstractPersistentDispatcherMultipleConsumers) {
+ AbstractPersistentDispatcherMultipleConsumers dispatcher =
+ (AbstractPersistentDispatcherMultipleConsumers) sub.getDispatcher();
+ cs.subscriptionHavePendingRead = dispatcher.isHavePendingRead();
+ cs.subscriptionHavePendingReplayRead = dispatcher.isHavePendingReplayRead();
} else if (sub.getDispatcher() instanceof PersistentDispatcherSingleActiveConsumer) {
PersistentDispatcherSingleActiveConsumer dispatcher =
(PersistentDispatcherSingleActiveConsumer) sub.getDispatcher();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/BrokerTestUtil.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/BrokerTestUtil.java
index ffcc3bf0881..7ed4542b250 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/BrokerTestUtil.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/BrokerTestUtil.java
@@ -36,6 +36,7 @@ import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BiConsumer;
import java.util.function.BiFunction;
+import java.util.stream.Stream;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.api.Consumer;
@@ -207,14 +208,28 @@ public class BrokerTestUtil {
* Receive messages concurrently from multiple consumers and handles them using the provided message handler.
*
* @param messageHandler the message handler
- * @param quietTimeout the duration of quiet time after which the method will stop waiting for more messages
- * @param consumers the consumers to receive messages from
- * @param <T> the message value type
+ * @param quietTimeout the duration of quiet time after which the method will stop waiting for more messages
+ * @param consumers the consumers to receive messages from
+ * @param <T> the message value type
+ */
+ public static <T> void receiveMessagesInThreads(BiFunction<Consumer<T>, Message<T>, Boolean> messageHandler,
+ final Duration quietTimeout,
+ Consumer<T>... consumers) {
+ receiveMessagesInThreads(messageHandler, quietTimeout, Arrays.stream(consumers).sequential());
+ }
+
+ /**
+ * Receive messages concurrently from multiple consumers and handles them using the provided message handler.
+ *
+ * @param messageHandler the message handler
+ * @param quietTimeout the duration of quiet time after which the method will stop waiting for more messages
+ * @param consumers the consumers to receive messages from
+ * @param <T> the message value type
*/
public static <T> void receiveMessagesInThreads(BiFunction<Consumer<T>, Message<T>, Boolean> messageHandler,
final Duration quietTimeout,
- Consumer<T>... consumers) {
- FutureUtil.waitForAll(Arrays.stream(consumers).sequential().map(consumer -> {
+ Stream<Consumer<T>> consumers) {
+ FutureUtil.waitForAll(consumers.map(consumer -> {
return CompletableFuture.runAsync(() -> {
try {
while (!Thread.currentThread().isInterrupted()) {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/AbstractDeliveryTrackerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/AbstractDeliveryTrackerTest.java
index ea6ffa2d70d..cdd0be58b34 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/AbstractDeliveryTrackerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/AbstractDeliveryTrackerTest.java
@@ -38,7 +38,7 @@ import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.bookkeeper.mledger.Position;
-import org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers;
+import org.apache.pulsar.broker.service.persistent.AbstractPersistentDispatcherMultipleConsumers;
import org.awaitility.Awaitility;
import org.testng.annotations.AfterClass;
import org.testng.annotations.Test;
@@ -49,7 +49,7 @@ public abstract class AbstractDeliveryTrackerTest {
protected final Timer timer =
new HashedWheelTimer(new DefaultThreadFactory("pulsar-in-memory-delayed-delivery-test"),
500, TimeUnit.MILLISECONDS);
- protected PersistentDispatcherMultipleConsumers dispatcher;
+ protected AbstractPersistentDispatcherMultipleConsumers dispatcher;
protected Clock clock;
protected AtomicLong clockTime;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/DelayedDeliveryTrackerFactoryTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/DelayedDeliveryTrackerFactoryTest.java
index 9861ab57237..ff6bf534129 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/DelayedDeliveryTrackerFactoryTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/DelayedDeliveryTrackerFactoryTest.java
@@ -24,6 +24,7 @@ import org.apache.pulsar.broker.delayed.bucket.RecoverDelayedDeliveryTrackerExce
import org.apache.pulsar.broker.service.BrokerService;
import org.apache.pulsar.broker.service.Dispatcher;
import org.apache.pulsar.broker.service.Subscription;
+import org.apache.pulsar.broker.service.persistent.AbstractPersistentDispatcherMultipleConsumers;
import org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers;
import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
@@ -67,10 +68,10 @@ public class DelayedDeliveryTrackerFactoryTest extends ProducerConsumerBase {
@Test
public void testFallbackToInMemoryTracker() throws Exception {
- Pair<BrokerService, PersistentDispatcherMultipleConsumers> pair =
+ Pair<BrokerService, AbstractPersistentDispatcherMultipleConsumers> pair =
mockDelayedDeliveryTrackerFactoryAndDispatcher();
BrokerService brokerService = pair.getLeft();
- PersistentDispatcherMultipleConsumers dispatcher = pair.getRight();
+ AbstractPersistentDispatcherMultipleConsumers dispatcher = pair.getRight();
// Since Mocked BucketDelayedDeliveryTrackerFactory.newTracker0() throws RecoverDelayedDeliveryTrackerException,
// the factory should be fallback to InMemoryDelayedDeliveryTrackerFactory
@@ -83,12 +84,13 @@ public class DelayedDeliveryTrackerFactoryTest extends ProducerConsumerBase {
}
- private Pair<BrokerService, PersistentDispatcherMultipleConsumers> mockDelayedDeliveryTrackerFactoryAndDispatcher()
+ private Pair<BrokerService, AbstractPersistentDispatcherMultipleConsumers> mockDelayedDeliveryTrackerFactoryAndDispatcher()
throws Exception {
BrokerService brokerService = Mockito.spy(pulsar.getBrokerService());
// Mock dispatcher
- PersistentDispatcherMultipleConsumers dispatcher = Mockito.mock(PersistentDispatcherMultipleConsumers.class);
+ AbstractPersistentDispatcherMultipleConsumers dispatcher =
+ Mockito.mock(AbstractPersistentDispatcherMultipleConsumers.class);
Mockito.doReturn("test").when(dispatcher).getName();
// Mock BucketDelayedDeliveryTrackerFactory
@Cleanup
@@ -113,10 +115,10 @@ public class DelayedDeliveryTrackerFactoryTest extends ProducerConsumerBase {
@Test
public void testFallbackToInMemoryTrackerFactoryFailed() throws Exception {
- Pair<BrokerService, PersistentDispatcherMultipleConsumers> pair =
+ Pair<BrokerService, AbstractPersistentDispatcherMultipleConsumers> pair =
mockDelayedDeliveryTrackerFactoryAndDispatcher();
BrokerService brokerService = pair.getLeft();
- PersistentDispatcherMultipleConsumers dispatcher = pair.getRight();
+ AbstractPersistentDispatcherMultipleConsumers dispatcher = pair.getRight();
// Mock InMemoryDelayedDeliveryTrackerFactory
@Cleanup
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/InMemoryDeliveryTrackerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/InMemoryDeliveryTrackerTest.java
index 6711aed924c..ff7763927d8 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/InMemoryDeliveryTrackerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/InMemoryDeliveryTrackerTest.java
@@ -37,7 +37,7 @@ import java.util.NavigableMap;
import java.util.TreeMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
-import org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers;
+import org.apache.pulsar.broker.service.persistent.AbstractPersistentDispatcherMultipleConsumers;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
@@ -46,7 +46,7 @@ public class InMemoryDeliveryTrackerTest extends AbstractDeliveryTrackerTest {
@DataProvider(name = "delayedTracker")
public Object[][] provider(Method method) throws Exception {
- dispatcher = mock(PersistentDispatcherMultipleConsumers.class);
+ dispatcher = mock(AbstractPersistentDispatcherMultipleConsumers.class);
clock = mock(Clock.class);
clockTime = new AtomicLong();
when(clock.millis()).then(x -> clockTime.get());
@@ -212,7 +212,8 @@ public class InMemoryDeliveryTrackerTest extends AbstractDeliveryTrackerTest {
Timer timer = new HashedWheelTimer(new DefaultThreadFactory("pulsar-in-memory-delayed-delivery-test"),
1, TimeUnit.MILLISECONDS);
- PersistentDispatcherMultipleConsumers dispatcher = mock(PersistentDispatcherMultipleConsumers.class);
+ AbstractPersistentDispatcherMultipleConsumers dispatcher =
+ mock(AbstractPersistentDispatcherMultipleConsumers.class);
AtomicLong clockTime = new AtomicLong();
Clock clock = mock(Clock.class);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTrackerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTrackerTest.java
index bf5a282a4ee..426bd50c96b 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTrackerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTrackerTest.java
@@ -51,7 +51,7 @@ import org.apache.commons.lang3.mutable.MutableLong;
import org.apache.pulsar.broker.delayed.AbstractDeliveryTrackerTest;
import org.apache.pulsar.broker.delayed.MockBucketSnapshotStorage;
import org.apache.pulsar.broker.delayed.MockManagedCursor;
-import org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers;
+import org.apache.pulsar.broker.service.persistent.AbstractPersistentDispatcherMultipleConsumers;
import org.awaitility.Awaitility;
import org.roaringbitmap.RoaringBitmap;
import org.roaringbitmap.buffer.ImmutableRoaringBitmap;
@@ -74,7 +74,7 @@ public class BucketDelayedDeliveryTrackerTest extends AbstractDeliveryTrackerTes
@DataProvider(name = "delayedTracker")
public Object[][] provider(Method method) throws Exception {
- dispatcher = mock(PersistentDispatcherMultipleConsumers.class);
+ dispatcher = mock(AbstractPersistentDispatcherMultipleConsumers.class);
clock = mock(Clock.class);
clockTime = new AtomicLong();
when(clock.millis()).then(x -> clockTime.get());
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BatchMessageTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BatchMessageTest.java
index c66eff2c8a1..2fd288239e3 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BatchMessageTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BatchMessageTest.java
@@ -37,7 +37,7 @@ import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import lombok.Cleanup;
-import org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers;
+import org.apache.pulsar.broker.service.persistent.AbstractPersistentDispatcherMultipleConsumers;
import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.client.api.BatcherBuilder;
@@ -786,7 +786,7 @@ public class BatchMessageTest extends BrokerTestBase {
}
latch.await();
- PersistentDispatcherMultipleConsumers dispatcher = (PersistentDispatcherMultipleConsumers) topic
+ AbstractPersistentDispatcherMultipleConsumers dispatcher = (AbstractPersistentDispatcherMultipleConsumers) topic
.getSubscription(subscriptionName).getDispatcher();
// check strategically to let ack-message receive by broker
retryStrategically((test) -> dispatcher.getConsumers().get(0).getUnackedMessages() == 0, 50, 150);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BatchMessageWithBatchIndexLevelTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BatchMessageWithBatchIndexLevelTest.java
index ed7f6974dd2..21a843a3efc 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BatchMessageWithBatchIndexLevelTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BatchMessageWithBatchIndexLevelTest.java
@@ -41,7 +41,7 @@ import org.apache.bookkeeper.mledger.Entry;
import org.apache.bookkeeper.mledger.impl.AckSetStateUtil;
import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
import org.apache.pulsar.broker.BrokerTestUtil;
-import org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers;
+import org.apache.pulsar.broker.service.persistent.AbstractPersistentDispatcherMultipleConsumers;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.ConsumerBuilder;
@@ -106,7 +106,7 @@ public class BatchMessageWithBatchIndexLevelTest extends BatchMessageTest {
}
FutureUtil.waitForAll(sendFutureList).get();
PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getTopicReference(topicName).get();
- PersistentDispatcherMultipleConsumers dispatcher = (PersistentDispatcherMultipleConsumers) topic
+ AbstractPersistentDispatcherMultipleConsumers dispatcher = (AbstractPersistentDispatcherMultipleConsumers) topic
.getSubscription(subscriptionName).getDispatcher();
Message<byte[]> receive1 = consumer.receive();
Message<byte[]> receive2 = consumer.receive();
@@ -515,8 +515,8 @@ public class BatchMessageWithBatchIndexLevelTest extends BatchMessageTest {
private ManagedCursorImpl getCursor(String topic, String sub) {
PersistentTopic persistentTopic =
(PersistentTopic) pulsar.getBrokerService().getTopic(topic, false).join().get();
- PersistentDispatcherMultipleConsumers dispatcher =
- (PersistentDispatcherMultipleConsumers) persistentTopic.getSubscription(sub).getDispatcher();
+ AbstractPersistentDispatcherMultipleConsumers dispatcher =
+ (AbstractPersistentDispatcherMultipleConsumers) persistentTopic.getSubscription(sub).getDispatcher();
return (ManagedCursorImpl) dispatcher.getCursor();
}
@@ -528,8 +528,8 @@ public class BatchMessageWithBatchIndexLevelTest extends BatchMessageTest {
CompletableFuture<Void> signal) throws Exception {
PersistentTopic persistentTopic =
(PersistentTopic) pulsar.getBrokerService().getTopic(topic, false).join().get();
- PersistentDispatcherMultipleConsumers dispatcher =
- (PersistentDispatcherMultipleConsumers) persistentTopic.getSubscription(sub).getDispatcher();
+ AbstractPersistentDispatcherMultipleConsumers dispatcher =
+ (AbstractPersistentDispatcherMultipleConsumers) persistentTopic.getSubscription(sub).getDispatcher();
org.apache.pulsar.broker.service.Consumer serviceConsumer = null;
for (org.apache.pulsar.broker.service.Consumer c : dispatcher.getConsumers()){
if (c.consumerName().equals(consumerName)) {
@@ -664,8 +664,8 @@ public class BatchMessageWithBatchIndexLevelTest extends BatchMessageTest {
private org.apache.pulsar.broker.service.Consumer getTheUniqueServiceConsumer(String topic, String sub) {
PersistentTopic persistentTopic =
(PersistentTopic) pulsar.getBrokerService(). getTopic(topic, false).join().get();
- PersistentDispatcherMultipleConsumers dispatcher =
- (PersistentDispatcherMultipleConsumers) persistentTopic.getSubscription(sub).getDispatcher();
+ AbstractPersistentDispatcherMultipleConsumers dispatcher =
+ (AbstractPersistentDispatcherMultipleConsumers) persistentTopic.getSubscription(sub).getDispatcher();
return dispatcher.getConsumers().iterator().next();
}
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentDispatcherFailoverConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentDispatcherFailoverConsumerTest.java
index 69f3e2e4d39..2899e9f2d67 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentDispatcherFailoverConsumerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentDispatcherFailoverConsumerTest.java
@@ -66,6 +66,7 @@ import org.apache.bookkeeper.mledger.PositionFactory;
import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.namespace.NamespaceService;
+import org.apache.pulsar.broker.service.persistent.AbstractPersistentDispatcherMultipleConsumers;
import org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers;
import org.apache.pulsar.broker.service.persistent.PersistentDispatcherSingleActiveConsumer;
import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
@@ -526,7 +527,8 @@ public class PersistentDispatcherFailoverConsumerTest {
PersistentTopic topic =
new PersistentTopic(successTopicName, ledgerMock, pulsarTestContext.getBrokerService());
- PersistentDispatcherMultipleConsumers dispatcher = new PersistentDispatcherMultipleConsumers(topic, cursorMock, null);
+ AbstractPersistentDispatcherMultipleConsumers
+ dispatcher = new PersistentDispatcherMultipleConsumers(topic, cursorMock, null);
Consumer consumer1 = createConsumer(topic, 0, 2, false, 1);
Consumer consumer2 = createConsumer(topic, 0, 2, false, 2);
Consumer consumer3 = createConsumer(topic, 0, 2, false, 3);
@@ -571,7 +573,8 @@ public class PersistentDispatcherFailoverConsumerTest {
public void testFewBlockedConsumerSamePriority() throws Exception{
PersistentTopic topic =
new PersistentTopic(successTopicName, ledgerMock, pulsarTestContext.getBrokerService());
- PersistentDispatcherMultipleConsumers dispatcher = new PersistentDispatcherMultipleConsumers(topic, cursorMock, null);
+ AbstractPersistentDispatcherMultipleConsumers
+ dispatcher = new PersistentDispatcherMultipleConsumers(topic, cursorMock, null);
Consumer consumer1 = createConsumer(topic, 0, 2, false, 1);
Consumer consumer2 = createConsumer(topic, 0, 2, false, 2);
Consumer consumer3 = createConsumer(topic, 0, 2, false, 3);
@@ -599,7 +602,8 @@ public class PersistentDispatcherFailoverConsumerTest {
public void testFewBlockedConsumerDifferentPriority() throws Exception {
PersistentTopic topic =
new PersistentTopic(successTopicName, ledgerMock, pulsarTestContext.getBrokerService());
- PersistentDispatcherMultipleConsumers dispatcher = new PersistentDispatcherMultipleConsumers(topic, cursorMock, null);
+ AbstractPersistentDispatcherMultipleConsumers
+ dispatcher = new PersistentDispatcherMultipleConsumers(topic, cursorMock, null);
Consumer consumer1 = createConsumer(topic, 0, 2, false, 1);
Consumer consumer2 = createConsumer(topic, 0, 2, false, 2);
Consumer consumer3 = createConsumer(topic, 0, 2, false, 3);
@@ -654,7 +658,8 @@ public class PersistentDispatcherFailoverConsumerTest {
public void testFewBlockedConsumerDifferentPriority2() throws Exception {
PersistentTopic topic =
new PersistentTopic(successTopicName, ledgerMock, pulsarTestContext.getBrokerService());
- PersistentDispatcherMultipleConsumers dispatcher = new PersistentDispatcherMultipleConsumers(topic, cursorMock, null);
+ AbstractPersistentDispatcherMultipleConsumers
+ dispatcher = new PersistentDispatcherMultipleConsumers(topic, cursorMock, null);
Consumer consumer1 = createConsumer(topic, 0, 2, true, 1);
Consumer consumer2 = createConsumer(topic, 0, 2, true, 2);
Consumer consumer3 = createConsumer(topic, 0, 2, true, 3);
@@ -677,7 +682,7 @@ public class PersistentDispatcherFailoverConsumerTest {
}
@SuppressWarnings("unchecked")
- private Consumer getNextConsumer(PersistentDispatcherMultipleConsumers dispatcher) throws Exception {
+ private Consumer getNextConsumer(AbstractPersistentDispatcherMultipleConsumers dispatcher) throws Exception {
Consumer consumer = dispatcher.getNextConsumer();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java
index 1e96da737dd..92b767104f6 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java
@@ -99,6 +99,7 @@ import org.apache.bookkeeper.test.MockedBookKeeperTestCase;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.namespace.NamespaceService;
+import org.apache.pulsar.broker.service.persistent.AbstractPersistentDispatcherMultipleConsumers;
import org.apache.pulsar.broker.service.persistent.GeoPersistentReplicator;
import org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers;
import org.apache.pulsar.broker.service.persistent.PersistentDispatcherSingleActiveConsumer;
@@ -346,7 +347,7 @@ public class PersistentTopicTest extends MockedBookKeeperTestCase {
when(cursor.getName()).thenReturn("cursor");
Subscription subscription = mock(Subscription.class);
when(subscription.getName()).thenReturn("sub");
- PersistentDispatcherMultipleConsumers dispatcher =
+ AbstractPersistentDispatcherMultipleConsumers dispatcher =
new PersistentDispatcherMultipleConsumers(topic, cursor, subscription);
dispatcher.readEntriesFailed(new ManagedLedgerException.InvalidCursorPositionException("failed"), null);
verify(topic, atLeast(1)).getBrokerService();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/BucketDelayedDeliveryTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/BucketDelayedDeliveryTest.java
index 20ea33fb3e1..4c8e6897df3 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/BucketDelayedDeliveryTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/BucketDelayedDeliveryTest.java
@@ -102,8 +102,8 @@ public class BucketDelayedDeliveryTest extends DelayedDeliveryTest {
Dispatcher dispatcher = pulsar.getBrokerService().getTopicReference(topic).get().getSubscription("sub").getDispatcher();
Awaitility.await().untilAsserted(() -> Assert.assertEquals(dispatcher.getNumberOfDelayedMessages(), 1000));
List<String> bucketKeys =
- ((PersistentDispatcherMultipleConsumers) dispatcher).getCursor().getCursorProperties().keySet().stream()
- .filter(x -> x.startsWith(CURSOR_INTERNAL_PROPERTY_PREFIX)).toList();
+ ((AbstractPersistentDispatcherMultipleConsumers) dispatcher).getCursor().getCursorProperties().keySet()
+ .stream().filter(x -> x.startsWith(CURSOR_INTERNAL_PROPERTY_PREFIX)).toList();
c1.close();
@@ -117,8 +117,8 @@ public class BucketDelayedDeliveryTest extends DelayedDeliveryTest {
Dispatcher dispatcher2 = pulsar.getBrokerService().getTopicReference(topic).get().getSubscription("sub").getDispatcher();
List<String> bucketKeys2 =
- ((PersistentDispatcherMultipleConsumers) dispatcher2).getCursor().getCursorProperties().keySet().stream()
- .filter(x -> x.startsWith(CURSOR_INTERNAL_PROPERTY_PREFIX)).toList();
+ ((AbstractPersistentDispatcherMultipleConsumers) dispatcher2).getCursor().getCursorProperties().keySet()
+ .stream().filter(x -> x.startsWith(CURSOR_INTERNAL_PROPERTY_PREFIX)).toList();
Awaitility.await().untilAsserted(() -> Assert.assertEquals(dispatcher2.getNumberOfDelayedMessages(), 1000));
Assert.assertEquals(bucketKeys, bucketKeys2);
@@ -152,7 +152,7 @@ public class BucketDelayedDeliveryTest extends DelayedDeliveryTest {
Awaitility.await().untilAsserted(() -> Assert.assertEquals(dispatcher.getNumberOfDelayedMessages(), 1000));
Map<String, String> cursorProperties =
- ((PersistentDispatcherMultipleConsumers) dispatcher).getCursor().getCursorProperties();
+ ((AbstractPersistentDispatcherMultipleConsumers) dispatcher).getCursor().getCursorProperties();
List<Long> bucketIds = cursorProperties.entrySet().stream()
.filter(x -> x.getKey().startsWith(CURSOR_INTERNAL_PROPERTY_PREFIX + "delayed.bucket")).map(
x -> Long.valueOf(x.getValue())).toList();
@@ -339,7 +339,7 @@ public class BucketDelayedDeliveryTest extends DelayedDeliveryTest {
Awaitility.await().untilAsserted(() -> Assert.assertEquals(dispatcher.getNumberOfDelayedMessages(), 1000));
Map<String, String> cursorProperties =
- ((PersistentDispatcherMultipleConsumers) dispatcher).getCursor().getCursorProperties();
+ ((AbstractPersistentDispatcherMultipleConsumers) dispatcher).getCursor().getCursorProperties();
List<Long> bucketIds = cursorProperties.entrySet().stream()
.filter(x -> x.getKey().startsWith(CURSOR_INTERNAL_PROPERTY_PREFIX + "delayed.bucket")).map(
x -> Long.valueOf(x.getValue())).toList();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/DelayedDeliveryTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/DelayedDeliveryTest.java
index 3ca966d2108..e47857e8ec6 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/DelayedDeliveryTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/DelayedDeliveryTest.java
@@ -259,12 +259,14 @@ public class DelayedDeliveryTest extends ProducerConsumerBase {
.subscribe();
// Simulate race condition with high frequency of calls to dispatcher.readMoreEntries()
- PersistentDispatcherMultipleConsumers d = (PersistentDispatcherMultipleConsumers) ((PersistentTopic) pulsar
- .getBrokerService().getTopicReference(topic).get()).getSubscription("shared-sub").getDispatcher();
+ AbstractPersistentDispatcherMultipleConsumers d =
+ (AbstractPersistentDispatcherMultipleConsumers) ((PersistentTopic) pulsar
+ .getBrokerService().getTopicReference(topic).get()).getSubscription("shared-sub")
+ .getDispatcher();
Thread t = new Thread(() -> {
while (true) {
synchronized (d) {
- d.readMoreEntries();
+ d.readMoreEntriesAsync();
}
try {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumersTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumersClassicTest.java
similarity index 94%
copy from pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumersTest.java
copy to pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumersClassicTest.java
index 052c5ceb5cd..487d99891fd 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumersTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumersClassicTest.java
@@ -44,7 +44,7 @@ import org.testng.annotations.Test;
@Slf4j
@Test(groups = "broker-api")
-public class PersistentDispatcherMultipleConsumersTest extends ProducerConsumerBase {
+public class PersistentDispatcherMultipleConsumersClassicTest extends ProducerConsumerBase {
@BeforeClass(alwaysRun = true)
@Override
@@ -131,8 +131,8 @@ public class PersistentDispatcherMultipleConsumersTest extends ProducerConsumerB
Subscription sub = Mockito.mock(PersistentSubscription.class);
Mockito.doReturn(topic).when(sub).getTopic();
// Mock the dispatcher.
- PersistentDispatcherMultipleConsumers dispatcher =
- Mockito.spy(new PersistentDispatcherMultipleConsumers(topic, cursor, sub));
+ PersistentDispatcherMultipleConsumersClassic dispatcher =
+ Mockito.spy(new PersistentDispatcherMultipleConsumersClassic(topic, cursor, sub));
// Return 10 permits to make the dispatcher can read more entries.
Mockito.doReturn(10).when(dispatcher).getFirstAvailableConsumerPermits();
@@ -154,12 +154,12 @@ public class PersistentDispatcherMultipleConsumersTest extends ProducerConsumerB
// Mock the readEntriesOrWait(...) to simulate the cursor is closed.
Mockito.doAnswer(inv -> {
- PersistentDispatcherMultipleConsumers dispatcher1 = inv.getArgument(2);
+ PersistentDispatcherMultipleConsumersClassic dispatcher1 = inv.getArgument(2);
dispatcher1.readEntriesFailed(new ManagedLedgerException.CursorAlreadyClosedException("cursor closed"),
null);
return null;
- }).when(cursor).asyncReadEntriesWithSkipOrWait(Mockito.anyInt(), Mockito.anyLong(), Mockito.eq(dispatcher),
- Mockito.any(), Mockito.any(), Mockito.any());
+ }).when(cursor).asyncReadEntriesOrWait(Mockito.anyInt(), Mockito.anyLong(), Mockito.eq(dispatcher),
+ Mockito.any(), Mockito.any());
dispatcher.readMoreEntries();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumersTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumersTest.java
index 052c5ceb5cd..772b1843d28 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumersTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumersTest.java
@@ -154,7 +154,7 @@ public class PersistentDispatcherMultipleConsumersTest extends ProducerConsumerB
// Mock the readEntriesOrWait(...) to simulate the cursor is closed.
Mockito.doAnswer(inv -> {
- PersistentDispatcherMultipleConsumers dispatcher1 = inv.getArgument(2);
+ AbstractPersistentDispatcherMultipleConsumers dispatcher1 = inv.getArgument(2);
dispatcher1.readEntriesFailed(new ManagedLedgerException.CursorAlreadyClosedException("cursor closed"),
null);
return null;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersClassicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersClassicTest.java
new file mode 100644
index 00000000000..1f40fd46aa3
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersClassicTest.java
@@ -0,0 +1,482 @@
+/*
+ * 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.nio.charset.StandardCharsets.UTF_8;
+import static org.apache.pulsar.common.protocol.Commands.serializeMetadataAndPayload;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.anyBoolean;
+import static org.mockito.Mockito.anyInt;
+import static org.mockito.Mockito.anyList;
+import static org.mockito.Mockito.anyLong;
+import static org.mockito.Mockito.anySet;
+import static org.mockito.Mockito.argThat;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.fail;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelPromise;
+import io.netty.channel.EventLoopGroup;
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+import org.apache.bookkeeper.common.util.OrderedExecutor;
+import org.apache.bookkeeper.mledger.Entry;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.PositionFactory;
+import org.apache.bookkeeper.mledger.impl.EntryImpl;
+import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
+import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.service.BrokerService;
+import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.EntryBatchIndexesAcks;
+import org.apache.pulsar.broker.service.EntryBatchSizes;
+import org.apache.pulsar.broker.service.RedeliveryTracker;
+import org.apache.pulsar.broker.service.plugin.EntryFilterProvider;
+import org.apache.pulsar.common.api.proto.KeySharedMeta;
+import org.apache.pulsar.common.api.proto.KeySharedMode;
+import org.apache.pulsar.common.api.proto.MessageMetadata;
+import org.apache.pulsar.common.policies.data.HierarchyTopicPolicies;
+import org.apache.pulsar.common.protocol.Commands;
+import org.apache.pulsar.common.protocol.Markers;
+import org.awaitility.Awaitility;
+import org.mockito.ArgumentCaptor;
+import org.testng.Assert;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+@Test(groups = "broker")
+public class PersistentStickyKeyDispatcherMultipleConsumersClassicTest {
+
+ private PulsarService pulsarMock;
+ private BrokerService brokerMock;
+ private ManagedCursorImpl cursorMock;
+ private Consumer consumerMock;
+ private PersistentTopic topicMock;
+ private PersistentSubscription subscriptionMock;
+ private ServiceConfiguration configMock;
+ private ChannelPromise channelMock;
+ private OrderedExecutor orderedExecutor;
+
+ private PersistentStickyKeyDispatcherMultipleConsumersClassic persistentDispatcher;
+
+ final String topicName = "persistent://public/default/testTopic";
+ final String subscriptionName = "testSubscription";
+
+ @BeforeMethod
+ public void setup() throws Exception {
+ configMock = mock(ServiceConfiguration.class);
+ doReturn(true).when(configMock).isSubscriptionRedeliveryTrackerEnabled();
+ doReturn(100).when(configMock).getDispatcherMaxReadBatchSize();
+ doReturn(true).when(configMock).isSubscriptionKeySharedUseConsistentHashing();
+ doReturn(1).when(configMock).getSubscriptionKeySharedConsistentHashingReplicaPoints();
+ doReturn(true).when(configMock).isDispatcherDispatchMessagesInSubscriptionThread();
+ doReturn(false).when(configMock).isAllowOverrideEntryFilters();
+
+ pulsarMock = mock(PulsarService.class);
+ doReturn(configMock).when(pulsarMock).getConfiguration();
+
+ EntryFilterProvider mockEntryFilterProvider = mock(EntryFilterProvider.class);
+ when(mockEntryFilterProvider.getBrokerEntryFilters()).thenReturn(Collections.emptyList());
+
+ brokerMock = mock(BrokerService.class);
+ doReturn(pulsarMock).when(brokerMock).pulsar();
+ when(brokerMock.getEntryFilterProvider()).thenReturn(mockEntryFilterProvider);
+
+ HierarchyTopicPolicies topicPolicies = new HierarchyTopicPolicies();
+ topicPolicies.getMaxConsumersPerSubscription().updateBrokerValue(0);
+
+ orderedExecutor = OrderedExecutor.newBuilder().build();
+ doReturn(orderedExecutor).when(brokerMock).getTopicOrderedExecutor();
+
+ EventLoopGroup eventLoopGroup = mock(EventLoopGroup.class);
+ doReturn(eventLoopGroup).when(brokerMock).executor();
+ doAnswer(invocation -> {
+ orderedExecutor.execute(((Runnable)invocation.getArguments()[0]));
+ return null;
+ }).when(eventLoopGroup).execute(any(Runnable.class));
+
+ topicMock = mock(PersistentTopic.class);
+ doReturn(brokerMock).when(topicMock).getBrokerService();
+ doReturn(topicName).when(topicMock).getName();
+ doReturn(topicPolicies).when(topicMock).getHierarchyTopicPolicies();
+
+ cursorMock = mock(ManagedCursorImpl.class);
+ doReturn(null).when(cursorMock).getLastIndividualDeletedRange();
+ doReturn(subscriptionName).when(cursorMock).getName();
+
+ consumerMock = mock(Consumer.class);
+ channelMock = mock(ChannelPromise.class);
+ doReturn("consumer1").when(consumerMock).consumerName();
+ doReturn(1000).when(consumerMock).getAvailablePermits();
+ doReturn(true).when(consumerMock).isWritable();
+ doReturn(channelMock).when(consumerMock).sendMessages(
+ anyList(),
+ any(EntryBatchSizes.class),
+ any(EntryBatchIndexesAcks.class),
+ anyInt(),
+ anyLong(),
+ anyLong(),
+ any(RedeliveryTracker.class)
+ );
+
+ subscriptionMock = mock(PersistentSubscription.class);
+ when(subscriptionMock.getTopic()).thenReturn(topicMock);
+ persistentDispatcher = new PersistentStickyKeyDispatcherMultipleConsumersClassic(
+ topicMock, cursorMock, subscriptionMock, configMock,
+ new KeySharedMeta().setKeySharedMode(KeySharedMode.AUTO_SPLIT));
+ }
+
+ @AfterMethod(alwaysRun = true)
+ public void cleanup() {
+ if (persistentDispatcher != null && !persistentDispatcher.isClosed()) {
+ persistentDispatcher.close();
+ }
+ if (orderedExecutor != null) {
+ orderedExecutor.shutdownNow();
+ orderedExecutor = null;
+ }
+ }
+
+ @Test(timeOut = 10000)
+ public void testAddConsumerWhenClosed() throws Exception {
+ persistentDispatcher.close().get();
+ Consumer consumer = mock(Consumer.class);
+ persistentDispatcher.addConsumer(consumer);
+ verify(consumer, times(1)).disconnect();
+ assertEquals(0, persistentDispatcher.getConsumers().size());
+ assertTrue(persistentDispatcher.getSelector().getConsumerKeyHashRanges().isEmpty());
+ }
+
+ @Test
+ public void testSendMarkerMessage() {
+ try {
+ persistentDispatcher.addConsumer(consumerMock);
+ persistentDispatcher.consumerFlow(consumerMock, 1000);
+ } catch (Exception e) {
+ fail("Failed to add mock consumer", e);
+ }
+
+ List<Entry> entries = new ArrayList<>();
+ ByteBuf markerMessage = Markers.newReplicatedSubscriptionsSnapshotRequest("testSnapshotId", "testSourceCluster");
+ entries.add(EntryImpl.create(1, 1, markerMessage));
+ entries.add(EntryImpl.create(1, 2, createMessage("message1", 1)));
+ entries.add(EntryImpl.create(1, 3, createMessage("message2", 2)));
+ entries.add(EntryImpl.create(1, 4, createMessage("message3", 3)));
+ entries.add(EntryImpl.create(1, 5, createMessage("message4", 4)));
+ entries.add(EntryImpl.create(1, 6, createMessage("message5", 5)));
+
+ try {
+ persistentDispatcher.readEntriesComplete(entries, PersistentStickyKeyDispatcherMultipleConsumersClassic.ReadType.Normal);
+ } catch (Exception e) {
+ fail("Failed to readEntriesComplete.", e);
+ }
+
+ Awaitility.await().untilAsserted(() -> {
+ ArgumentCaptor<Integer> totalMessagesCaptor = ArgumentCaptor.forClass(Integer.class);
+ verify(consumerMock, times(1)).sendMessages(
+ anyList(),
+ any(EntryBatchSizes.class),
+ any(EntryBatchIndexesAcks.class),
+ totalMessagesCaptor.capture(),
+ anyLong(),
+ anyLong(),
+ any(RedeliveryTracker.class)
+ );
+
+ List<Integer> allTotalMessagesCaptor = totalMessagesCaptor.getAllValues();
+ Assert.assertEquals(allTotalMessagesCaptor.get(0).intValue(), 5);
+ });
+ }
+
+ @Test(timeOut = 10000)
+ public void testSendMessage() {
+ KeySharedMeta keySharedMeta = new KeySharedMeta().setKeySharedMode(KeySharedMode.STICKY);
+ PersistentStickyKeyDispatcherMultipleConsumersClassic
+ persistentDispatcher = new PersistentStickyKeyDispatcherMultipleConsumersClassic(
+ topicMock, cursorMock, subscriptionMock, configMock, keySharedMeta);
+ try {
+ keySharedMeta.addHashRange()
+ .setStart(0)
+ .setEnd(9);
+
+ Consumer consumerMock = mock(Consumer.class);
+ doReturn(keySharedMeta).when(consumerMock).getKeySharedMeta();
+ persistentDispatcher.addConsumer(consumerMock);
+ persistentDispatcher.consumerFlow(consumerMock, 1000);
+ } catch (Exception e) {
+ fail("Failed to add mock consumer", e);
+ }
+
+ List<Entry> entries = new ArrayList<>();
+ entries.add(EntryImpl.create(1, 1, createMessage("message1", 1)));
+ entries.add(EntryImpl.create(1, 2, createMessage("message2", 2)));
+
+ try {
+ //Should success,see issue #8960
+ persistentDispatcher.readEntriesComplete(entries, PersistentStickyKeyDispatcherMultipleConsumersClassic.ReadType.Normal);
+ } catch (Exception e) {
+ fail("Failed to readEntriesComplete.", e);
+ }
+ }
+
+ @Test
+ public void testSkipRedeliverTemporally() {
+ final Consumer slowConsumerMock = mock(Consumer.class);
+ final ChannelPromise slowChannelMock = mock(ChannelPromise.class);
+ // add entries to redeliver and read target
+ final List<Entry> redeliverEntries = new ArrayList<>();
+ redeliverEntries.add(EntryImpl.create(1, 1, createMessage("message1", 1, "key1")));
+ final List<Entry> readEntries = new ArrayList<>();
+ readEntries.add(EntryImpl.create(1, 2, createMessage("message2", 2, "key1")));
+ readEntries.add(EntryImpl.create(1, 3, createMessage("message3", 3, "key2")));
+
+ try {
+ Field totalAvailablePermitsField = PersistentDispatcherMultipleConsumersClassic.class.getDeclaredField("totalAvailablePermits");
+ totalAvailablePermitsField.setAccessible(true);
+ totalAvailablePermitsField.set(persistentDispatcher, 1000);
+
+ doAnswer(invocationOnMock -> {
+ ((PersistentStickyKeyDispatcherMultipleConsumersClassic) invocationOnMock.getArgument(2))
+ .readEntriesComplete(readEntries, PersistentStickyKeyDispatcherMultipleConsumersClassic.ReadType.Normal);
+ return null;
+ }).when(cursorMock).asyncReadEntriesOrWait(
+ anyInt(), anyLong(), any(PersistentStickyKeyDispatcherMultipleConsumersClassic.class),
+ eq(PersistentStickyKeyDispatcherMultipleConsumersClassic.ReadType.Normal), any());
+ } catch (Exception e) {
+ fail("Failed to set to field", e);
+ }
+
+ // Create 2Consumers
+ try {
+ doReturn("consumer2").when(slowConsumerMock).consumerName();
+ // Change slowConsumer availablePermits to 0 and back to normal
+ when(slowConsumerMock.getAvailablePermits())
+ .thenReturn(0)
+ .thenReturn(1);
+ doReturn(true).when(slowConsumerMock).isWritable();
+ doReturn(slowChannelMock).when(slowConsumerMock).sendMessages(
+ anyList(),
+ any(EntryBatchSizes.class),
+ any(EntryBatchIndexesAcks.class),
+ anyInt(),
+ anyLong(),
+ anyLong(),
+ any(RedeliveryTracker.class)
+ );
+
+ persistentDispatcher.addConsumer(consumerMock);
+ persistentDispatcher.addConsumer(slowConsumerMock);
+ } catch (Exception e) {
+ fail("Failed to add mock consumer", e);
+ }
+
+ // run PersistentStickyKeyDispatcherMultipleConsumers#sendMessagesToConsumers
+ // run readMoreEntries internally (and skip internally)
+ // Change slowConsumer availablePermits to 1
+ // run PersistentStickyKeyDispatcherMultipleConsumers#sendMessagesToConsumers internally
+ // and then stop to dispatch to slowConsumer
+ if (persistentDispatcher.sendMessagesToConsumers(PersistentStickyKeyDispatcherMultipleConsumersClassic.ReadType.Normal,
+ redeliverEntries, true)) {
+ persistentDispatcher.readMoreEntriesAsync();
+ }
+
+ Awaitility.await().untilAsserted(() -> {
+ verify(consumerMock, times(1)).sendMessages(
+ argThat(arg -> {
+ assertEquals(arg.size(), 1);
+ Entry entry = arg.get(0);
+ assertEquals(entry.getLedgerId(), 1);
+ assertEquals(entry.getEntryId(), 3);
+ return true;
+ }),
+ any(EntryBatchSizes.class),
+ any(EntryBatchIndexesAcks.class),
+ anyInt(),
+ anyLong(),
+ anyLong(),
+ any(RedeliveryTracker.class)
+ );
+ });
+ verify(slowConsumerMock, times(0)).sendMessages(
+ anyList(),
+ any(EntryBatchSizes.class),
+ any(EntryBatchIndexesAcks.class),
+ anyInt(),
+ anyLong(),
+ anyLong(),
+ any(RedeliveryTracker.class)
+ );
+ }
+
+ @Test(timeOut = 30000)
+ public void testMessageRedelivery() throws Exception {
+ final Queue<Position> actualEntriesToConsumer1 = new ConcurrentLinkedQueue<>();
+ final Queue<Position> actualEntriesToConsumer2 = new ConcurrentLinkedQueue<>();
+
+ final Queue<Position> expectedEntriesToConsumer1 = new ConcurrentLinkedQueue<>();
+ expectedEntriesToConsumer1.add(PositionFactory.create(1, 1));
+ final Queue<Position> expectedEntriesToConsumer2 = new ConcurrentLinkedQueue<>();
+ expectedEntriesToConsumer2.add(PositionFactory.create(1, 2));
+ expectedEntriesToConsumer2.add(PositionFactory.create(1, 3));
+
+ final AtomicInteger remainingEntriesNum = new AtomicInteger(
+ expectedEntriesToConsumer1.size() + expectedEntriesToConsumer2.size());
+
+ // Messages with key1 are routed to consumer1 and messages with key2 are routed to consumer2
+ final List<Entry> allEntries = new ArrayList<>();
+ allEntries.add(EntryImpl.create(1, 1, createMessage("message1", 1, "key2")));
+ allEntries.add(EntryImpl.create(1, 2, createMessage("message2", 2, "key1")));
+ allEntries.add(EntryImpl.create(1, 3, createMessage("message3", 3, "key1")));
+ allEntries.forEach(entry -> ((EntryImpl) entry).retain());
+
+ final List<Entry> redeliverEntries = new ArrayList<>();
+ redeliverEntries.add(allEntries.get(0)); // message1
+ final List<Entry> readEntries = new ArrayList<>();
+ readEntries.add(allEntries.get(2)); // message3
+
+ final Consumer consumer1 = mock(Consumer.class);
+ doReturn("consumer1").when(consumer1).consumerName();
+ // Change availablePermits of consumer1 to 0 and then back to normal
+ when(consumer1.getAvailablePermits()).thenReturn(0).thenReturn(10);
+ doReturn(true).when(consumer1).isWritable();
+ doAnswer(invocationOnMock -> {
+ @SuppressWarnings("unchecked")
+ List<Entry> entries = (List<Entry>) invocationOnMock.getArgument(0);
+ for (Entry entry : entries) {
+ remainingEntriesNum.decrementAndGet();
+ actualEntriesToConsumer1.add(entry.getPosition());
+ }
+ return channelMock;
+ }).when(consumer1).sendMessages(anyList(), any(EntryBatchSizes.class), any(EntryBatchIndexesAcks.class),
+ anyInt(), anyLong(), anyLong(), any(RedeliveryTracker.class));
+
+ final Consumer consumer2 = mock(Consumer.class);
+ doReturn("consumer2").when(consumer2).consumerName();
+ when(consumer2.getAvailablePermits()).thenReturn(10);
+ doReturn(true).when(consumer2).isWritable();
+ doAnswer(invocationOnMock -> {
+ @SuppressWarnings("unchecked")
+ List<Entry> entries = (List<Entry>) invocationOnMock.getArgument(0);
+ for (Entry entry : entries) {
+ remainingEntriesNum.decrementAndGet();
+ actualEntriesToConsumer2.add(entry.getPosition());
+ }
+ return channelMock;
+ }).when(consumer2).sendMessages(anyList(), any(EntryBatchSizes.class), any(EntryBatchIndexesAcks.class),
+ anyInt(), anyLong(), anyLong(), any(RedeliveryTracker.class));
+
+ persistentDispatcher.addConsumer(consumer1);
+ persistentDispatcher.addConsumer(consumer2);
+
+ final Field totalAvailablePermitsField = PersistentDispatcherMultipleConsumersClassic.class
+ .getDeclaredField("totalAvailablePermits");
+ totalAvailablePermitsField.setAccessible(true);
+ totalAvailablePermitsField.set(persistentDispatcher, 1000);
+
+ final Field redeliveryMessagesField = PersistentDispatcherMultipleConsumersClassic.class
+ .getDeclaredField("redeliveryMessages");
+ redeliveryMessagesField.setAccessible(true);
+ MessageRedeliveryController redeliveryMessages = (MessageRedeliveryController) redeliveryMessagesField
+ .get(persistentDispatcher);
+ redeliveryMessages.add(allEntries.get(0).getLedgerId(), allEntries.get(0).getEntryId(),
+ persistentDispatcher.getStickyKeyHash(allEntries.get(0))); // message1
+ redeliveryMessages.add(allEntries.get(1).getLedgerId(), allEntries.get(1).getEntryId(),
+ persistentDispatcher.getStickyKeyHash(allEntries.get(1))); // message2
+
+ // Mock Cursor#asyncReplayEntries
+ doAnswer(invocationOnMock -> {
+ @SuppressWarnings("unchecked")
+ Set<Position> positions = (Set<Position>) invocationOnMock.getArgument(0);
+ List<Entry> entries = allEntries.stream().filter(entry -> positions.contains(entry.getPosition()))
+ .collect(Collectors.toList());
+ if (!entries.isEmpty()) {
+ ((PersistentStickyKeyDispatcherMultipleConsumersClassic) invocationOnMock.getArgument(1))
+ .readEntriesComplete(entries, PersistentStickyKeyDispatcherMultipleConsumersClassic.ReadType.Replay);
+ }
+ return Collections.emptySet();
+ }).when(cursorMock).asyncReplayEntries(anySet(), any(PersistentStickyKeyDispatcherMultipleConsumersClassic.class),
+ eq(PersistentStickyKeyDispatcherMultipleConsumersClassic.ReadType.Replay), anyBoolean());
+
+ // Mock Cursor#asyncReadEntriesOrWait
+ AtomicBoolean asyncReadEntriesOrWaitCalled = new AtomicBoolean();
+ doAnswer(invocationOnMock -> {
+ if (asyncReadEntriesOrWaitCalled.compareAndSet(false, true)) {
+ ((PersistentStickyKeyDispatcherMultipleConsumersClassic) invocationOnMock.getArgument(2))
+ .readEntriesComplete(readEntries, PersistentStickyKeyDispatcherMultipleConsumersClassic.ReadType.Normal);
+ } else {
+ ((PersistentStickyKeyDispatcherMultipleConsumersClassic) invocationOnMock.getArgument(2))
+ .readEntriesComplete(Collections.emptyList(), PersistentStickyKeyDispatcherMultipleConsumersClassic.ReadType.Normal);
+ }
+ return null;
+ }).when(cursorMock).asyncReadEntriesOrWait(anyInt(), anyLong(),
+ any(PersistentStickyKeyDispatcherMultipleConsumersClassic.class),
+ eq(PersistentStickyKeyDispatcherMultipleConsumersClassic.ReadType.Normal), any());
+
+ // (1) Run sendMessagesToConsumers
+ // (2) Attempts to send message1 to consumer1 but skipped because availablePermits is 0
+ // (3) Change availablePermits of consumer1 to 10
+ // (4) Run readMoreEntries internally
+ // (5) Run sendMessagesToConsumers internally
+ // (6) Attempts to send message3 to consumer2 but skipped because redeliveryMessages contains message2
+ persistentDispatcher.sendMessagesToConsumers(PersistentStickyKeyDispatcherMultipleConsumersClassic.ReadType.Replay,
+ redeliverEntries, true);
+ while (remainingEntriesNum.get() > 0) {
+ // (7) Run readMoreEntries and resend message1 to consumer1 and message2-3 to consumer2
+ persistentDispatcher.readMoreEntries();
+ }
+
+ assertThat(actualEntriesToConsumer1).containsExactlyElementsOf(expectedEntriesToConsumer1);
+ assertThat(actualEntriesToConsumer2).containsExactlyElementsOf(expectedEntriesToConsumer2);
+
+ allEntries.forEach(entry -> entry.release());
+ }
+
+ private ByteBuf createMessage(String message, int sequenceId) {
+ return createMessage(message, sequenceId, "testKey");
+ }
+
+ private ByteBuf createMessage(String message, int sequenceId, String key) {
+ MessageMetadata messageMetadata = new MessageMetadata()
+ .setSequenceId(sequenceId)
+ .setProducerName("testProducer")
+ .setPartitionKey(key)
+ .setPartitionKeyB64Encoded(false)
+ .setPublishTime(System.currentTimeMillis());
+ return serializeMetadataAndPayload(Commands.ChecksumType.Crc32c, messageMetadata, Unpooled.copiedBuffer(message.getBytes(UTF_8)));
+ }
+}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersTest.java
index 4b29ead984e..7234f0caefc 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersTest.java
@@ -720,7 +720,7 @@ public class PersistentStickyKeyDispatcherMultipleConsumersTest {
AtomicInteger reScheduleReadInMsCalled = new AtomicInteger(0);
AtomicBoolean delayAllMessages = new AtomicBoolean(true);
- PersistentDispatcherMultipleConsumers dispatcher;
+ AbstractPersistentDispatcherMultipleConsumers dispatcher;
if (isKeyShared) {
dispatcher = new PersistentStickyKeyDispatcherMultipleConsumers(
topicMock, cursorMock, subscriptionMock, configMock,
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/ConsumerStatsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/ConsumerStatsTest.java
index 14403765105..5b2998216e8 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/ConsumerStatsTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/ConsumerStatsTest.java
@@ -233,7 +233,7 @@ public class ConsumerStatsTest extends ProducerConsumerBase {
"unackedMessages",
"avgMessagesPerEntry",
"blockedConsumerOnUnackedMsgs",
- "lastSentPositionWhenJoining",
+ "readPositionWhenJoining",
"lastAckedTime",
"lastAckedTimestamp",
"lastConsumedTime",
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/KeySharedSubscriptionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/KeySharedSubscriptionTest.java
index 2b16647f559..08efb6d9583 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/KeySharedSubscriptionTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/KeySharedSubscriptionTest.java
@@ -65,10 +65,13 @@ import org.apache.bookkeeper.mledger.PositionFactory;
import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
import org.apache.commons.lang3.tuple.Pair;
+import org.apache.pulsar.broker.BrokerTestUtil;
import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.service.Dispatcher;
import org.apache.pulsar.broker.service.DrainingHashesTracker;
import org.apache.pulsar.broker.service.PendingAcksMap;
import org.apache.pulsar.broker.service.StickyKeyConsumerSelector;
+import org.apache.pulsar.broker.service.StickyKeyDispatcher;
import org.apache.pulsar.broker.service.Topic;
import org.apache.pulsar.broker.service.nonpersistent.NonPersistentStickyKeyDispatcherMultipleConsumers;
import org.apache.pulsar.broker.service.persistent.PersistentStickyKeyDispatcherMultipleConsumers;
@@ -79,6 +82,7 @@ import org.apache.pulsar.common.api.proto.KeySharedMode;
import org.apache.pulsar.common.naming.TopicDomain;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.schema.KeyValue;
+import org.apache.pulsar.tests.KeySharedImplementationType;
import org.assertj.core.api.Assertions;
import org.awaitility.Awaitility;
import org.mockito.Mockito;
@@ -89,53 +93,80 @@ import org.testng.annotations.AfterClass;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.DataProvider;
+import org.testng.annotations.Factory;
import org.testng.annotations.Test;
@Test(groups = "broker-impl")
public class KeySharedSubscriptionTest extends ProducerConsumerBase {
-
private static final Logger log = LoggerFactory.getLogger(KeySharedSubscriptionTest.class);
private static final List<String> keys = Arrays.asList("0", "1", "2", "3", "4", "5", "6", "7", "8", "9");
private static final String SUBSCRIPTION_NAME = "key_shared";
+ private final KeySharedImplementationType implementationType;
+
+ // Comment out the next line (Factory annotation) to run tests manually in IntelliJ, one-by-one
+ @Factory
+ public static Object[] createTestInstances() {
+ return KeySharedImplementationType.generateTestInstances(KeySharedSubscriptionTest::new);
+ }
+
+ public KeySharedSubscriptionTest() {
+ // set the default implementation type for manual running in IntelliJ
+ this(KeySharedImplementationType.DEFAULT);
+ }
+
+ public KeySharedSubscriptionTest(KeySharedImplementationType implementationType) {
+ this.implementationType = implementationType;
+ }
+
+ private Object[][] prependImplementationTypeToData(Object[][] data) {
+ return implementationType.prependImplementationTypeToData(data);
+ }
+
+ @DataProvider(name = "currentImplementationType")
+ public Object[] currentImplementationType() {
+ return new Object[]{ implementationType };
+ }
@DataProvider(name = "batch")
- public Object[] batchProvider() {
- return new Object[] {
- false,
- true
- };
+ public Object[][] batchProvider() {
+ return prependImplementationTypeToData(new Object[][]{
+ {false},
+ {true}
+ });
}
@DataProvider(name = "partitioned")
public Object[][] partitionedProvider() {
- return new Object[][] {
- { false },
- { true }
- };
+ return prependImplementationTypeToData(new Object[][]{
+ {false},
+ {true}
+ });
}
@DataProvider(name = "data")
public Object[][] dataProvider() {
- return new Object[][] {
+ return prependImplementationTypeToData(new Object[][]{
// Topic-Type and "Batching"
- { "persistent", false },
- { "persistent", true },
- { "non-persistent", false },
- { "non-persistent", true },
- };
+ {"persistent", false},
+ {"persistent", true},
+ {"non-persistent", false},
+ {"non-persistent", true},
+ });
}
@DataProvider(name = "topicDomain")
public Object[][] topicDomainProvider() {
- return new Object[][] {
- { "persistent" },
- { "non-persistent" }
- };
+ return prependImplementationTypeToData(new Object[][]{
+ {"persistent"},
+ {"non-persistent"}
+ });
}
@BeforeClass(alwaysRun = true)
@Override
protected void setup() throws Exception {
+ conf.setSubscriptionKeySharedUseClassicPersistentImplementation(implementationType.classic);
+ conf.setSubscriptionSharedUseClassicPersistentImplementation(implementationType.classic);
this.conf.setUnblockStuckSubscriptionEnabled(true);
super.internalSetup();
super.producerBaseSetup();
@@ -170,7 +201,8 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase {
private static final int NUMBER_OF_KEYS = 300;
@Test(dataProvider = "data")
- public void testSendAndReceiveWithHashRangeAutoSplitStickyKeyConsumerSelector(String topicType, boolean enableBatch)
+ public void testSendAndReceiveWithHashRangeAutoSplitStickyKeyConsumerSelector(KeySharedImplementationType impl,
+ String topicType, boolean enableBatch)
throws PulsarClientException {
String topic = topicType + "://public/default/key_shared-" + UUID.randomUUID();
@@ -197,7 +229,7 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase {
}
@Test(dataProvider = "data")
- public void testSendAndReceiveWithBatching(String topicType, boolean enableBatch) throws Exception {
+ public void testSendAndReceiveWithBatching(KeySharedImplementationType impl, String topicType, boolean enableBatch) throws Exception {
String topic = topicType + "://public/default/key_shared-" + UUID.randomUUID();
@Cleanup
@@ -242,7 +274,9 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase {
}
@Test(dataProvider = "batch")
- public void testSendAndReceiveWithHashRangeExclusiveStickyKeyConsumerSelector(boolean enableBatch) throws PulsarClientException {
+ public void testSendAndReceiveWithHashRangeExclusiveStickyKeyConsumerSelector(KeySharedImplementationType impl,
+ boolean enableBatch)
+ throws PulsarClientException {
String topic = "persistent://public/default/key_shared_exclusive-" + UUID.randomUUID();
@Cleanup
@@ -294,8 +328,9 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase {
@Test(dataProvider = "data")
public void testConsumerCrashSendAndReceiveWithHashRangeAutoSplitStickyKeyConsumerSelector(
- String topicType,
- boolean enableBatch
+ KeySharedImplementationType impl,
+ String topicType,
+ boolean enableBatch
) throws PulsarClientException, InterruptedException {
String topic = topicType + "://public/default/key_shared_consumer_crash-" + UUID.randomUUID();
@@ -338,8 +373,9 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase {
@Test(dataProvider = "data")
public void testNoKeySendAndReceiveWithHashRangeAutoSplitStickyKeyConsumerSelector(
- String topicType,
- boolean enableBatch
+ KeySharedImplementationType impl,
+ String topicType,
+ boolean enableBatch
) throws PulsarClientException {
String topic = topicType + "://public/default/key_shared_no_key-" + UUID.randomUUID();
@@ -365,7 +401,8 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase {
}
@Test(dataProvider = "batch")
- public void testNoKeySendAndReceiveWithHashRangeExclusiveStickyKeyConsumerSelector(boolean enableBatch)
+ public void testNoKeySendAndReceiveWithHashRangeExclusiveStickyKeyConsumerSelector(KeySharedImplementationType impl,
+ boolean enableBatch)
throws PulsarClientException {
String topic = "persistent://public/default/key_shared_no_key_exclusive-" + UUID.randomUUID();
@@ -415,7 +452,8 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase {
}
@Test(dataProvider = "batch")
- public void testOrderingKeyWithHashRangeAutoSplitStickyKeyConsumerSelector(boolean enableBatch)
+ public void testOrderingKeyWithHashRangeAutoSplitStickyKeyConsumerSelector(KeySharedImplementationType impl,
+ boolean enableBatch)
throws PulsarClientException {
String topic = "persistent://public/default/key_shared_ordering_key-" + UUID.randomUUID();
@@ -443,7 +481,8 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase {
}
@Test(dataProvider = "batch")
- public void testOrderingKeyWithHashRangeExclusiveStickyKeyConsumerSelector(boolean enableBatch)
+ public void testOrderingKeyWithHashRangeExclusiveStickyKeyConsumerSelector(KeySharedImplementationType impl,
+ boolean enableBatch)
throws PulsarClientException {
String topic = "persistent://public/default/key_shared_exclusive_ordering_key-" + UUID.randomUUID();
@@ -512,8 +551,8 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase {
}
}
- @Test
- public void testCannotUseAcknowledgeCumulative() throws PulsarClientException {
+ @Test(dataProvider = "currentImplementationType")
+ public void testCannotUseAcknowledgeCumulative(KeySharedImplementationType impl) throws PulsarClientException {
String topic = "persistent://public/default/key_shared_ack_cumulative-" + UUID.randomUUID();
@Cleanup
@@ -538,7 +577,7 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase {
}
@Test(dataProvider = "batch")
- public void testMakingProgressWithSlowerConsumer(boolean enableBatch) throws Exception {
+ public void testMakingProgressWithSlowerConsumer(KeySharedImplementationType impl, boolean enableBatch) throws Exception {
String topic = "testMakingProgressWithSlowerConsumer-" + UUID.randomUUID();
String slowKey = "slowKey";
@@ -620,8 +659,8 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase {
}
}
- @Test
- public void testOrderingWhenAddingConsumers() throws Exception {
+ @Test(dataProvider = "currentImplementationType")
+ public void testOrderingWhenAddingConsumers(KeySharedImplementationType impl) throws Exception {
String topic = "testOrderingWhenAddingConsumers-" + UUID.randomUUID();
int numberOfKeys = 10;
@@ -668,13 +707,13 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase {
}
@SneakyThrows
- private PersistentStickyKeyDispatcherMultipleConsumers getDispatcher(String topic, String subscription) {
- return (PersistentStickyKeyDispatcherMultipleConsumers) pulsar.getBrokerService().getTopicIfExists(topic).get()
+ private StickyKeyDispatcher getDispatcher(String topic, String subscription) {
+ return (StickyKeyDispatcher) pulsar.getBrokerService().getTopicIfExists(topic).get()
.get().getSubscription(subscription).getDispatcher();
}
- @Test
- public void testReadAheadWithConfiguredLookAheadLimit() throws Exception {
+ @Test(dataProvider = "currentImplementationType")
+ public void testReadAheadWithConfiguredLookAheadLimit(KeySharedImplementationType impl) throws Exception {
String topic = "testReadAheadWithConfiguredLookAheadLimit-" + UUID.randomUUID();
// Set the look ahead limit to 50 for subscriptions
@@ -730,8 +769,8 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase {
assertTrue(entryId < 100);
}
- @Test
- public void testRemoveFirstConsumer() throws Exception {
+ @Test(dataProvider = "currentImplementationType")
+ public void testRemoveFirstConsumer(KeySharedImplementationType impl) throws Exception {
String topic = "testReadAheadWhenAddingConsumers-" + UUID.randomUUID();
int numberOfKeys = 10;
@@ -788,8 +827,8 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase {
}
}
- @Test
- public void testHashRangeConflict() throws PulsarClientException {
+ @Test(dataProvider = "currentImplementationType")
+ public void testHashRangeConflict(KeySharedImplementationType impl) throws PulsarClientException {
final String topic = "persistent://public/default/testHashRangeConflict-" + UUID.randomUUID().toString();
final String sub = "test";
@@ -799,7 +838,7 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase {
Consumer<String> consumer2 = createFixedHashRangesConsumer(topic, sub, Range.of(100,399));
Assert.assertTrue(consumer2.isConnected());
- PersistentStickyKeyDispatcherMultipleConsumers dispatcher = getDispatcher(topic, sub);
+ StickyKeyDispatcher dispatcher = getDispatcher(topic, sub);
Assert.assertEquals(dispatcher.getConsumers().size(), 2);
try {
@@ -849,8 +888,8 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase {
Assert.assertFalse(dispatcher.isConsumerConnected());
}
- @Test
- public void testWithMessageCompression() throws Exception {
+ @Test(dataProvider = "currentImplementationType")
+ public void testWithMessageCompression(KeySharedImplementationType impl) throws Exception {
final String topic = "testWithMessageCompression" + UUID.randomUUID().toString();
Producer<byte[]> producer = pulsarClient.newProducer()
.topic(topic)
@@ -876,8 +915,8 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase {
consumer.close();
}
- @Test
- public void testAttachKeyToMessageMetadata() throws PulsarClientException {
+ @Test(dataProvider = "currentImplementationType")
+ public void testAttachKeyToMessageMetadata(KeySharedImplementationType impl) throws PulsarClientException {
String topic = "persistent://public/default/key_shared-" + UUID.randomUUID();
@Cleanup
@@ -904,8 +943,8 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase {
receiveAndCheckDistribution(Lists.newArrayList(consumer1, consumer2, consumer3), 1000);
}
- @Test
- public void testContinueDispatchMessagesWhenMessageTTL() throws Exception {
+ @Test(dataProvider = "currentImplementationType")
+ public void testContinueDispatchMessagesWhenMessageTTL(KeySharedImplementationType impl) throws Exception {
int defaultTTLSec = 3;
int totalMessages = 1000;
int numberOfKeys = 50;
@@ -922,7 +961,7 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase {
.subscriptionType(SubscriptionType.Key_Shared)
.subscribe();
- PersistentStickyKeyDispatcherMultipleConsumers dispatcher = getDispatcher(topic, subName);
+ StickyKeyDispatcher dispatcher = getDispatcher(topic, subName);
StickyKeyConsumerSelector selector = dispatcher.getSelector();
@Cleanup
@@ -964,7 +1003,7 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase {
if (received != null) {
int stickyKeyHash = selector.makeStickyKeyHash(received.getKeyBytes());
DrainingHashesTracker.DrainingHashEntry entry =
- dispatcher.getDrainingHashesTracker().getEntry(stickyKeyHash);
+ !impl.classic ? getDrainingHashesTracker(dispatcher).getEntry(stickyKeyHash) : null;
Assertions.fail("Received message %s with sticky key hash that should have been blocked: %d. entry=%s, "
+ "included in blockedHashes=%s",
received.getMessageId(), stickyKeyHash, entry, blockedHashes.contains(stickyKeyHash));
@@ -982,10 +1021,10 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase {
received = consumer3.receive(1, TimeUnit.SECONDS);
} catch (PulsarClientException ignore) {
}
- if (received != null) {
+ if (received != null && !impl.classic) {
int stickyKeyHash = selector.makeStickyKeyHash(received.getKeyBytes());
DrainingHashesTracker.DrainingHashEntry entry =
- dispatcher.getDrainingHashesTracker().getEntry(stickyKeyHash);
+ !impl.classic ? getDrainingHashesTracker(dispatcher).getEntry(stickyKeyHash) : null;
Assertions.fail("Received message %s with sticky key hash that should have been blocked: %d. entry=%s, "
+ "included in blockedHashes=%s",
received.getMessageId(), stickyKeyHash, entry, blockedHashes.contains(stickyKeyHash));
@@ -1018,8 +1057,12 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase {
count -> assertThat(count.get()).isGreaterThan(0));
}
+ private DrainingHashesTracker getDrainingHashesTracker(Dispatcher dispatcher) {
+ return ((PersistentStickyKeyDispatcherMultipleConsumers) dispatcher).getDrainingHashesTracker();
+ }
+
@Test(dataProvider = "partitioned")
- public void testOrderingWithConsumerListener(boolean partitioned) throws Exception {
+ public void testOrderingWithConsumerListener(KeySharedImplementationType impl, boolean partitioned) throws Exception {
final String topic = "persistent://public/default/key_shared-" + UUID.randomUUID();
if (partitioned) {
admin.topics().createPartitionedTopic(topic, 3);
@@ -1075,8 +1118,8 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase {
consumer.close();
}
- @Test
- public void testKeySharedConsumerWithEncrypted() throws Exception {
+ @Test(dataProvider = "currentImplementationType")
+ public void testKeySharedConsumerWithEncrypted(KeySharedImplementationType impl) throws Exception {
final String topic = "persistent://public/default/key_shared-" + UUID.randomUUID();
final int totalMessages = 100;
@@ -1142,7 +1185,7 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase {
}
@Test(dataProvider = "topicDomain")
- public void testSelectorChangedAfterAllConsumerDisconnected(String topicDomain) throws PulsarClientException,
+ public void testSelectorChangedAfterAllConsumerDisconnected(KeySharedImplementationType impl, String topicDomain) throws PulsarClientException,
ExecutionException, InterruptedException {
final String topicName = TopicName.get(topicDomain, "public", "default",
"testSelectorChangedAfterAllConsumerDisconnected" + UUID.randomUUID()).toString();
@@ -1187,8 +1230,8 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase {
consumer1.close();
}
- @Test
- public void testAllowOutOfOrderDeliveryChangedAfterAllConsumerDisconnected() throws Exception {
+ @Test(dataProvider = "currentImplementationType")
+ public void testAllowOutOfOrderDeliveryChangedAfterAllConsumerDisconnected(KeySharedImplementationType impl) throws Exception {
final String topicName = "persistent://public/default/change-allow-ooo-delivery-" + UUID.randomUUID();
final String subName = "my-sub";
@@ -1207,7 +1250,7 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase {
producer.send("message".getBytes());
Awaitility.await().untilAsserted(() -> assertNotNull(consumer1.receive(100, TimeUnit.MILLISECONDS)));
- PersistentStickyKeyDispatcherMultipleConsumers dispatcher = getDispatcher(topicName, subName);
+ StickyKeyDispatcher dispatcher = getDispatcher(topicName, subName);
assertTrue(dispatcher.isAllowOutOfOrderDelivery());
consumer1.close();
@@ -1225,8 +1268,8 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase {
consumer2.close();
}
- @Test(timeOut = 30_000)
- public void testCheckConsumersWithSameName() throws Exception {
+ @Test(timeOut = 30_000, dataProvider = "currentImplementationType")
+ public void testCheckConsumersWithSameName(KeySharedImplementationType impl) throws Exception {
final String topicName = "persistent://public/default/same-name-" + UUID.randomUUID();
final String subName = "my-sub";
final String consumerName = "name";
@@ -1270,25 +1313,37 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase {
@Cleanup("shutdownNow")
ExecutorService e = Executors.newCachedThreadPool();
e.submit(() -> {
- while (l.getCount() > 0) {
+ while (l.getCount() > 0 && !Thread.currentThread().isInterrupted()) {
try {
Message<String> msg = c2.receive(1, TimeUnit.SECONDS);
+ if (msg == null) {
+ continue;
+ }
c2.acknowledge(msg);
l.countDown();
} catch (PulsarClientException ex) {
ex.printStackTrace();
+ if (ex instanceof PulsarClientException.AlreadyClosedException) {
+ break;
+ }
}
}
});
e.submit(() -> {
- while (l.getCount() > 0) {
+ while (l.getCount() > 0 && !Thread.currentThread().isInterrupted()) {
try {
Message<String> msg = c3.receive(1, TimeUnit.SECONDS);
+ if (msg == null) {
+ continue;
+ }
c3.acknowledge(msg);
l.countDown();
} catch (PulsarClientException ex) {
ex.printStackTrace();
+ if (ex instanceof PulsarClientException.AlreadyClosedException) {
+ break;
+ }
}
}
});
@@ -1303,7 +1358,7 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase {
private KeySharedMode getKeySharedModeOfSubscription(Topic topic, String subscription) {
if (TopicName.get(topic.getName()).getDomain().equals(TopicDomain.persistent)) {
- return ((PersistentStickyKeyDispatcherMultipleConsumers) topic.getSubscription(subscription)
+ return ((StickyKeyDispatcher) topic.getSubscription(subscription)
.getDispatcher()).getKeySharedMode();
} else if (TopicName.get(topic.getName()).getDomain().equals(TopicDomain.non_persistent)) {
return ((NonPersistentStickyKeyDispatcherMultipleConsumers) topic.getSubscription(subscription)
@@ -1390,45 +1445,37 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase {
*/
private void receiveAndCheckDistribution(List<Consumer<?>> consumers, int expectedTotalMessage) throws PulsarClientException {
// Add a key so that we know this key was already assigned to one consumer
- Map<String, Consumer<?>> keyToConsumer = new HashMap<>();
- Map<Consumer<?>, Integer> messagesPerConsumer = new HashMap<>();
+ Map<String, Consumer<?>> keyToConsumer = new ConcurrentHashMap<>();
+ Map<Consumer<?>, AtomicInteger> messagesPerConsumer = new ConcurrentHashMap<>();
+ AtomicInteger totalMessages = new AtomicInteger();
- int totalMessages = 0;
+ BiFunction<Consumer<Object>, Message<Object>, Boolean> messageHandler = (consumer, msg) -> {
+ totalMessages.incrementAndGet();
+ messagesPerConsumer.computeIfAbsent(consumer, k -> new AtomicInteger()).incrementAndGet();
+ try {
+ consumer.acknowledge(msg);
+ } catch (PulsarClientException e) {
+ throw new RuntimeException(e);
+ }
- for (Consumer<?> c : consumers) {
- int messagesForThisConsumer = 0;
- while (true) {
- Message<?> msg = c.receive(100, TimeUnit.MILLISECONDS);
- if (msg == null) {
- // Go to next consumer
- messagesPerConsumer.put(c, messagesForThisConsumer);
- break;
- }
-
- ++totalMessages;
- ++messagesForThisConsumer;
- c.acknowledge(msg);
-
- if (msg.hasKey() || msg.hasOrderingKey()) {
- String key = msg.hasOrderingKey() ? new String(msg.getOrderingKey()) : msg.getKey();
- Consumer<?> assignedConsumer = keyToConsumer.get(key);
- if (assignedConsumer == null) {
- // This is a new key
- keyToConsumer.put(key, c);
- } else {
- // The consumer should be the same
- assertEquals(c, assignedConsumer);
- }
+ if (msg.hasKey() || msg.hasOrderingKey()) {
+ String key = msg.hasOrderingKey() ? new String(msg.getOrderingKey()) : msg.getKey();
+ Consumer<?> assignedConsumer = keyToConsumer.putIfAbsent(key, consumer);
+ if (assignedConsumer != null && !assignedConsumer.equals(consumer)) {
+ assertEquals(consumer, assignedConsumer);
}
}
- }
+ return true;
+ };
- final double PERCENT_ERROR = 0.40; // 40 %
+ BrokerTestUtil.receiveMessagesInThreads(messageHandler, Duration.ofMillis(250),
+ consumers.stream().map(Consumer.class::cast));
- double expectedMessagesPerConsumer = totalMessages / consumers.size();
- Assert.assertEquals(expectedTotalMessage, totalMessages);
- for (int count : messagesPerConsumer.values()) {
- Assert.assertEquals(count, expectedMessagesPerConsumer, expectedMessagesPerConsumer * PERCENT_ERROR);
+ final double PERCENT_ERROR = 0.40; // 40 %
+ double expectedMessagesPerConsumer = totalMessages.get() / (double) consumers.size();
+ Assert.assertEquals(expectedTotalMessage, totalMessages.get());
+ for (AtomicInteger count : messagesPerConsumer.values()) {
+ Assert.assertEquals(count.get(), expectedMessagesPerConsumer, expectedMessagesPerConsumer * PERCENT_ERROR);
}
}
@@ -1531,8 +1578,8 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase {
}
}
- @Test
- public void testStickyKeyRangesRestartConsumers() throws Exception {
+ @Test(dataProvider = "currentImplementationType")
+ public void testStickyKeyRangesRestartConsumers(KeySharedImplementationType impl) throws Exception {
final String topic = TopicName.get("persistent", "public", "default",
"testStickyKeyRangesRestartConsumers" + UUID.randomUUID()).toString();
@@ -1663,8 +1710,8 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase {
producerFuture.get();
}
- @Test
- public void testContinueDispatchMessagesWhenMessageDelayed() throws Exception {
+ @Test(dataProvider = "currentImplementationType")
+ public void testContinueDispatchMessagesWhenMessageDelayed(KeySharedImplementationType impl) throws Exception {
int delayedMessages = 40;
int messages = 40;
int sum = 0;
@@ -1765,8 +1812,8 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase {
return replyReadCounter;
}
- @Test
- public void testNoRepeatedReadAndDiscard() throws Exception {
+ @Test(dataProvider = "currentImplementationType")
+ public void testNoRepeatedReadAndDiscard(KeySharedImplementationType impl) throws Exception {
int delayedMessages = 100;
int numberOfKeys = delayedMessages;
final String topic = newUniqueName("persistent://public/default/tp");
@@ -1839,10 +1886,10 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase {
@DataProvider(name = "allowKeySharedOutOfOrder")
public Object[][] allowKeySharedOutOfOrder() {
- return new Object[][]{
+ return prependImplementationTypeToData(new Object[][]{
{true},
{false}
- };
+ });
}
/**
@@ -1860,7 +1907,7 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase {
* - at last, all messages will be received.
*/
@Test(timeOut = 180 * 1000, dataProvider = "allowKeySharedOutOfOrder") // the test will be finished in 60s.
- public void testRecentJoinedPosWillNotStuckOtherConsumer(boolean allowKeySharedOutOfOrder) throws Exception {
+ public void testRecentJoinedPosWillNotStuckOtherConsumer(KeySharedImplementationType impl, boolean allowKeySharedOutOfOrder) throws Exception {
final int messagesSentPerTime = 100;
final Set<Integer> totalReceivedMessages = new TreeSet<>();
final String topic = newUniqueName("persistent://public/default/tp");
@@ -2020,8 +2067,10 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase {
admin.topics().delete(topic, false);
}
- @Test
- public void testReadAheadLimit() throws Exception {
+ @Test(dataProvider = "currentImplementationType")
+ public void testReadAheadLimit(KeySharedImplementationType impl) throws Exception {
+ // skip for classic implementation since the feature is not implemented
+ impl.skipIfClassic();
String topic = "testReadAheadLimit-" + UUID.randomUUID();
int numberOfKeys = 1000;
long pauseTime = 100L;
@@ -2040,7 +2089,7 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase {
.subscribe()
.close();
- PersistentStickyKeyDispatcherMultipleConsumers dispatcher = getDispatcher(topic, subscriptionName);
+ StickyKeyDispatcher dispatcher = getDispatcher(topic, subscriptionName);
// create a function to use for checking the number of messages in replay
Runnable checkLimit = () -> {
@@ -2145,16 +2194,18 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase {
private StickyKeyConsumerSelector getSelector(String topic, String subscription) {
Topic t = pulsar.getBrokerService().getTopicIfExists(topic).get().get();
PersistentSubscription sub = (PersistentSubscription) t.getSubscription(subscription);
- PersistentStickyKeyDispatcherMultipleConsumers dispatcher =
- (PersistentStickyKeyDispatcherMultipleConsumers) sub.getDispatcher();
+ StickyKeyDispatcher dispatcher = (StickyKeyDispatcher) sub.getDispatcher();
return dispatcher.getSelector();
}
// This test case simulates a rolling restart scenario with behaviors that can trigger out-of-order issues.
// In earlier versions of Pulsar, this issue occurred in about 25% of cases.
// To increase the probability of reproducing the issue, use the invocationCount parameter.
- @Test//(invocationCount = 50)
- public void testOrderingAfterReconnects() throws Exception {
+ @Test(dataProvider = "currentImplementationType")//(invocationCount = 50)
+ public void testOrderingAfterReconnects(KeySharedImplementationType impl) throws Exception {
+ // skip for classic implementation since this fails
+ impl.skipIfClassic();
+
String topic = newUniqueName("testOrderingAfterReconnects");
int numberOfKeys = 1000;
long pauseTime = 100L;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SubscriptionMessageDispatchThrottlingTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SubscriptionMessageDispatchThrottlingTest.java
index 02de11a2bcc..ce554ab2d9c 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SubscriptionMessageDispatchThrottlingTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SubscriptionMessageDispatchThrottlingTest.java
@@ -25,8 +25,8 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.pulsar.broker.BrokerTestUtil;
import org.apache.pulsar.broker.service.Dispatcher;
+import org.apache.pulsar.broker.service.persistent.AbstractPersistentDispatcherMultipleConsumers;
import org.apache.pulsar.broker.service.persistent.DispatchRateLimiter;
-import org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers;
import org.apache.pulsar.broker.service.persistent.PersistentDispatcherSingleActiveConsumer;
import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
@@ -93,7 +93,7 @@ public class SubscriptionMessageDispatchThrottlingTest extends MessageDispatchTh
DispatchRateLimiter subRateLimiter = null;
Dispatcher subDispatcher = topic.getSubscription(subName).getDispatcher();
- if (subDispatcher instanceof PersistentDispatcherMultipleConsumers) {
+ if (subDispatcher instanceof AbstractPersistentDispatcherMultipleConsumers) {
subRateLimiter = subDispatcher.getRateLimiter().get();
} else if (subDispatcher instanceof PersistentDispatcherSingleActiveConsumer) {
subRateLimiter = subDispatcher.getRateLimiter().get();
@@ -180,7 +180,7 @@ public class SubscriptionMessageDispatchThrottlingTest extends MessageDispatchTh
DispatchRateLimiter subRateLimiter = null;
Dispatcher subDispatcher = topic.getSubscription(subName).getDispatcher();
- if (subDispatcher instanceof PersistentDispatcherMultipleConsumers) {
+ if (subDispatcher instanceof AbstractPersistentDispatcherMultipleConsumers) {
subRateLimiter = subDispatcher.getRateLimiter().get();
} else if (subDispatcher instanceof PersistentDispatcherSingleActiveConsumer) {
subRateLimiter = subDispatcher.getRateLimiter().get();
@@ -264,7 +264,7 @@ public class SubscriptionMessageDispatchThrottlingTest extends MessageDispatchTh
DispatchRateLimiter subRateLimiter = null;
Dispatcher subDispatcher = topic.getSubscription(subName).getDispatcher();
- if (subDispatcher instanceof PersistentDispatcherMultipleConsumers) {
+ if (subDispatcher instanceof AbstractPersistentDispatcherMultipleConsumers) {
subRateLimiter = subDispatcher.getRateLimiter().get();
} else if (subDispatcher instanceof PersistentDispatcherSingleActiveConsumer) {
subRateLimiter = subDispatcher.getRateLimiter().get();
@@ -356,7 +356,7 @@ public class SubscriptionMessageDispatchThrottlingTest extends MessageDispatchTh
DispatchRateLimiter subRateLimiter = null;
Dispatcher subDispatcher = topic.getSubscription(subName).getDispatcher();
- if (subDispatcher instanceof PersistentDispatcherMultipleConsumers) {
+ if (subDispatcher instanceof AbstractPersistentDispatcherMultipleConsumers) {
subRateLimiter = subDispatcher.getRateLimiter().get();
} else if (subDispatcher instanceof PersistentDispatcherSingleActiveConsumer) {
subRateLimiter = subDispatcher.getRateLimiter().get();
@@ -442,7 +442,7 @@ public class SubscriptionMessageDispatchThrottlingTest extends MessageDispatchTh
DispatchRateLimiter subRateLimiter = null;
Dispatcher subDispatcher = topic.getSubscription(subName).getDispatcher();
- if (subDispatcher instanceof PersistentDispatcherMultipleConsumers) {
+ if (subDispatcher instanceof AbstractPersistentDispatcherMultipleConsumers) {
subRateLimiter = subDispatcher.getRateLimiter().get();
} else if (subDispatcher instanceof PersistentDispatcherSingleActiveConsumer) {
subRateLimiter = subDispatcher.getRateLimiter().get();
@@ -649,7 +649,7 @@ public class SubscriptionMessageDispatchThrottlingTest extends MessageDispatchTh
DispatchRateLimiter subRateLimiter = null;
Dispatcher subDispatcher = topic.getSubscription(subName).getDispatcher();
- if (subDispatcher instanceof PersistentDispatcherMultipleConsumers) {
+ if (subDispatcher instanceof AbstractPersistentDispatcherMultipleConsumers) {
subRateLimiter = subDispatcher.getRateLimiter().get();
} else if (subDispatcher instanceof PersistentDispatcherSingleActiveConsumer) {
subRateLimiter = subDispatcher.getRateLimiter().get();
@@ -805,7 +805,7 @@ public class SubscriptionMessageDispatchThrottlingTest extends MessageDispatchTh
DispatchRateLimiter subRateLimiter = null;
Dispatcher subDispatcher = topic.getSubscription(subName1).getDispatcher();
- if (subDispatcher instanceof PersistentDispatcherMultipleConsumers) {
+ if (subDispatcher instanceof AbstractPersistentDispatcherMultipleConsumers) {
subRateLimiter = subDispatcher.getRateLimiter().get();
} else if (subDispatcher instanceof PersistentDispatcherSingleActiveConsumer) {
subRateLimiter = subDispatcher.getRateLimiter().get();
@@ -855,7 +855,7 @@ public class SubscriptionMessageDispatchThrottlingTest extends MessageDispatchTh
.subscribe();
subDispatcher = topic2.getSubscription(subName2).getDispatcher();
- if (subDispatcher instanceof PersistentDispatcherMultipleConsumers) {
+ if (subDispatcher instanceof AbstractPersistentDispatcherMultipleConsumers) {
subRateLimiter = subDispatcher.getRateLimiter().get();
} else if (subDispatcher instanceof PersistentDispatcherSingleActiveConsumer) {
subRateLimiter = subDispatcher.getRateLimiter().get();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SubscriptionPauseOnAckStatPersistTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SubscriptionPauseOnAckStatPersistTest.java
index 390e81ad664..8fe1f3e58d9 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SubscriptionPauseOnAckStatPersistTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SubscriptionPauseOnAckStatPersistTest.java
@@ -32,6 +32,7 @@ import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
import org.apache.pulsar.broker.BrokerTestUtil;
import org.apache.pulsar.broker.service.Dispatcher;
import org.apache.pulsar.broker.service.SystemTopicBasedTopicPoliciesService;
+import org.apache.pulsar.broker.service.persistent.AbstractPersistentDispatcherMultipleConsumers;
import org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers;
import org.apache.pulsar.broker.service.persistent.PersistentDispatcherSingleActiveConsumer;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
@@ -104,8 +105,8 @@ public class SubscriptionPauseOnAckStatPersistTest extends ProducerConsumerBase
PersistentTopic persistentTopic =
(PersistentTopic) pulsar.getBrokerService().getTopic(tpName, false).join().get();
Dispatcher dispatcher = persistentTopic.getSubscription(cursorName).getDispatcher();
- if (dispatcher instanceof PersistentDispatcherMultipleConsumers) {
- ((PersistentDispatcherMultipleConsumers) dispatcher).readMoreEntries();
+ if (dispatcher instanceof AbstractPersistentDispatcherMultipleConsumers) {
+ ((AbstractPersistentDispatcherMultipleConsumers) dispatcher).readMoreEntriesAsync();
} else if (dispatcher instanceof PersistentDispatcherSingleActiveConsumer) {
PersistentDispatcherSingleActiveConsumer persistentDispatcherSingleActiveConsumer =
((PersistentDispatcherSingleActiveConsumer) dispatcher);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/KeySharedSubscriptionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/KeySharedSubscriptionMaxUnackedMessagesTest.java
similarity index 85%
rename from pulsar-broker/src/test/java/org/apache/pulsar/client/impl/KeySharedSubscriptionTest.java
rename to pulsar-broker/src/test/java/org/apache/pulsar/client/impl/KeySharedSubscriptionMaxUnackedMessagesTest.java
index 704af89777f..b3ef641ca19 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/KeySharedSubscriptionTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/KeySharedSubscriptionMaxUnackedMessagesTest.java
@@ -34,8 +34,8 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import lombok.SneakyThrows;
import org.apache.pulsar.broker.service.StickyKeyConsumerSelector;
+import org.apache.pulsar.broker.service.StickyKeyDispatcher;
import org.apache.pulsar.broker.service.Topic;
-import org.apache.pulsar.broker.service.persistent.PersistentStickyKeyDispatcherMultipleConsumers;
import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
import org.apache.pulsar.client.api.BatcherBuilder;
import org.apache.pulsar.client.api.Consumer;
@@ -47,19 +47,39 @@ import org.apache.pulsar.client.api.ProducerConsumerBase;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Range;
import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.tests.KeySharedImplementationType;
import org.awaitility.Awaitility;
import org.testng.Assert;
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;
@Test(groups = "broker-impl")
-public class KeySharedSubscriptionTest extends ProducerConsumerBase {
+public class KeySharedSubscriptionMaxUnackedMessagesTest extends ProducerConsumerBase {
+ private final KeySharedImplementationType implementationType;
+
+ // Comment out the next line (Factory annotation) to run tests manually in IntelliJ, one-by-one
+ @Factory
+ public static Object[] createTestInstances() {
+ return KeySharedImplementationType.generateTestInstances(KeySharedSubscriptionMaxUnackedMessagesTest::new);
+ }
+
+ public KeySharedSubscriptionMaxUnackedMessagesTest() {
+ // set the default implementation type for manual running in IntelliJ
+ this(KeySharedImplementationType.DEFAULT);
+ }
+
+ public KeySharedSubscriptionMaxUnackedMessagesTest(KeySharedImplementationType implementationType) {
+ this.implementationType = implementationType;
+ }
@Override
@BeforeMethod
protected void setup() throws Exception {
+ conf.setSubscriptionKeySharedUseClassicPersistentImplementation(implementationType.classic);
+ conf.setSubscriptionSharedUseClassicPersistentImplementation(implementationType.classic);
conf.setMaxUnackedMessagesPerConsumer(10);
super.internalSetup();
super.producerBaseSetup();
@@ -82,16 +102,17 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase {
@DataProvider
public Object[][] subType() {
- return new Object[][] {
- { SubscriptionType.Shared, null },
- { SubscriptionType.Key_Shared, KeySharedSelectorType.AutoSplit_ConsistentHashing },
- { SubscriptionType.Key_Shared, KeySharedSelectorType.AutoSplit_Classic },
- { SubscriptionType.Key_Shared, KeySharedSelectorType.Sticky }
- };
+ return implementationType.prependImplementationTypeToData(new Object[][]{
+ {SubscriptionType.Shared, null},
+ {SubscriptionType.Key_Shared, KeySharedSelectorType.AutoSplit_ConsistentHashing},
+ {SubscriptionType.Key_Shared, KeySharedSelectorType.AutoSplit_Classic},
+ {SubscriptionType.Key_Shared, KeySharedSelectorType.Sticky}
+ });
}
@Test(dataProvider = "subType", timeOut = 30000)
- public void testCanRecoverConsumptionWhenLiftMaxUnAckedMessagesRestriction(SubscriptionType subscriptionType,
+ public void testCanRecoverConsumptionWhenLiftMaxUnAckedMessagesRestriction(KeySharedImplementationType impl,
+ SubscriptionType subscriptionType,
KeySharedSelectorType selectorType)
throws PulsarClientException {
if (selectorType == KeySharedSelectorType.AutoSplit_Classic) {
@@ -258,8 +279,7 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase {
private StickyKeyConsumerSelector getSelector(String topic, String subscription) {
Topic t = pulsar.getBrokerService().getTopicIfExists(topic).get().get();
PersistentSubscription sub = (PersistentSubscription) t.getSubscription(subscription);
- PersistentStickyKeyDispatcherMultipleConsumers dispatcher =
- (PersistentStickyKeyDispatcherMultipleConsumers) sub.getDispatcher();
+ StickyKeyDispatcher dispatcher = (StickyKeyDispatcher) sub.getDispatcher();
return dispatcher.getSelector();
}
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/tests/KeySharedImplementationType.java b/pulsar-broker/src/test/java/org/apache/pulsar/tests/KeySharedImplementationType.java
new file mode 100644
index 00000000000..39b504131fc
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/tests/KeySharedImplementationType.java
@@ -0,0 +1,61 @@
+/*
+ * 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.tests;
+
+import java.util.Arrays;
+import java.util.function.Function;
+import org.testng.SkipException;
+
+/**
+ * KeyShared implementation type used in test.
+ */
+public enum KeySharedImplementationType {
+ // default implementation, PIP-379
+ PIP379(false),
+ // classic implementation before PIP-282 and PIP-379
+ Classic(true);
+
+ public static final KeySharedImplementationType DEFAULT = PIP379;
+ public final boolean classic;
+
+ KeySharedImplementationType(boolean classic) {
+ this.classic = classic;
+ }
+
+ public void skipIfClassic() {
+ if (classic) {
+ throw new SkipException("Test is not applicable for classic implementation");
+ }
+ }
+
+ public Object[][] prependImplementationTypeToData(Object[][] data) {
+ return Arrays.stream(data)
+ .map(array -> {
+ Object[] newArray = new Object[array.length + 1];
+ newArray[0] = this;
+ System.arraycopy(array, 0, newArray, 1, array.length);
+ return newArray;
+ })
+ .toArray(Object[][]::new);
+ }
+
+ public static Object[] generateTestInstances(Function<KeySharedImplementationType, Object> testInstanceFactory) {
+ return Arrays.stream(KeySharedImplementationType.values()).map(testInstanceFactory).toArray();
+ }
+}
diff --git a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/ConsumerStats.java b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/ConsumerStats.java
index 5f2cf7b209e..d2d3600df96 100644
--- a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/ConsumerStats.java
+++ b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/ConsumerStats.java
@@ -72,8 +72,8 @@ public interface ConsumerStats {
/** Flag to verify if consumer is blocked due to reaching threshold of unacked messages. */
boolean isBlockedConsumerOnUnackedMsgs();
- /** The last sent position of the cursor when the consumer joining. */
- String getLastSentPositionWhenJoining();
+ /** The read position of the cursor when the consumer joining. */
+ String getReadPositionWhenJoining();
/** Address of this consumer. */
String getAddress();
diff --git a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/SubscriptionStats.java b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/SubscriptionStats.java
index 7b7c1f5765c..ce3a080a855 100644
--- a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/SubscriptionStats.java
+++ b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/SubscriptionStats.java
@@ -118,6 +118,9 @@ public interface SubscriptionStats {
/** Whether the Key_Shared subscription mode is AUTO_SPLIT or STICKY. */
String getKeySharedMode();
+ /** This is for Key_Shared subscription to get the recentJoinedConsumers in the Key_Shared subscription. */
+ Map<String, String> getConsumersAfterMarkDeletePosition();
+
/** SubscriptionProperties (key/value strings) associated with this subscribe. */
Map<String, String> getSubscriptionProperties();
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/ConsumerStatsImpl.java b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/ConsumerStatsImpl.java
index b4c5d21e692..de36b330b7f 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/ConsumerStatsImpl.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/ConsumerStatsImpl.java
@@ -77,8 +77,8 @@ public class ConsumerStatsImpl implements ConsumerStats {
/** Flag to verify if consumer is blocked due to reaching threshold of unacked messages. */
public boolean blockedConsumerOnUnackedMsgs;
- /** The last sent position of the cursor when the consumer joining. */
- public String lastSentPositionWhenJoining;
+ /** The read position of the cursor when the consumer joining. */
+ public String readPositionWhenJoining;
/** Address of this consumer. */
private String address;
@@ -113,7 +113,7 @@ public class ConsumerStatsImpl implements ConsumerStats {
this.availablePermits += stats.availablePermits;
this.unackedMessages += stats.unackedMessages;
this.blockedConsumerOnUnackedMsgs = stats.blockedConsumerOnUnackedMsgs;
- this.lastSentPositionWhenJoining = stats.lastSentPositionWhenJoining;
+ this.readPositionWhenJoining = stats.readPositionWhenJoining;
return this;
}
@@ -141,8 +141,8 @@ public class ConsumerStatsImpl implements ConsumerStats {
this.clientVersion = clientVersion;
}
- public String getLastSentPositionWhenJoining() {
- return lastSentPositionWhenJoining;
+ public String getReadPositionWhenJoining() {
+ return readPositionWhenJoining;
}
public String getLastAckedTime() {
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/SubscriptionStatsImpl.java b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/SubscriptionStatsImpl.java
index 4206a4aa8d6..12734a5586c 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/SubscriptionStatsImpl.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/SubscriptionStatsImpl.java
@@ -21,6 +21,7 @@ package org.apache.pulsar.common.policies.data.stats;
import com.fasterxml.jackson.annotation.JsonIgnore;
import java.util.ArrayList;
import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
@@ -125,8 +126,8 @@ public class SubscriptionStatsImpl implements SubscriptionStats {
/** Whether the Key_Shared subscription mode is AUTO_SPLIT or STICKY. */
public String keySharedMode;
- /** The last sent position of the cursor. This is for Key_Shared subscription. */
- public String lastSentPosition;
+ /** This is for Key_Shared subscription to get the recentJoinedConsumers in the Key_Shared subscription. */
+ public Map<String, String> consumersAfterMarkDeletePosition;
/** The number of non-contiguous deleted messages ranges. */
public int nonContiguousDeletedMessagesRanges;
@@ -153,6 +154,7 @@ public class SubscriptionStatsImpl implements SubscriptionStats {
public SubscriptionStatsImpl() {
this.consumers = new ArrayList<>();
+ this.consumersAfterMarkDeletePosition = new LinkedHashMap<>();
this.subscriptionProperties = new HashMap<>();
this.bucketDelayedIndexStats = new HashMap<>();
}
@@ -177,6 +179,7 @@ public class SubscriptionStatsImpl implements SubscriptionStats {
lastExpireTimestamp = 0L;
lastMarkDeleteAdvancedTimestamp = 0L;
consumers.clear();
+ consumersAfterMarkDeletePosition.clear();
nonContiguousDeletedMessagesRanges = 0;
nonContiguousDeletedMessagesRangesSerializedSize = 0;
earliestMsgPublishTimeInBacklog = 0L;
@@ -222,6 +225,7 @@ public class SubscriptionStatsImpl implements SubscriptionStats {
}
}
this.allowOutOfOrderDelivery |= stats.allowOutOfOrderDelivery;
+ this.consumersAfterMarkDeletePosition.putAll(stats.consumersAfterMarkDeletePosition);
this.nonContiguousDeletedMessagesRanges += stats.nonContiguousDeletedMessagesRanges;
this.nonContiguousDeletedMessagesRangesSerializedSize += stats.nonContiguousDeletedMessagesRangesSerializedSize;
if (this.earliestMsgPublishTimeInBacklog != 0 && stats.earliestMsgPublishTimeInBacklog != 0) {