[improve][broker] Add limits for Key_Shared Subscription look ahead in dispatching

conf/broker.conf

@@ -355,6 +355,25 @@ maxUnackedMessagesPerBroker=0
 # limit/2 messages
 maxUnackedMessagesPerSubscriptionOnBrokerBlocked=0.16
 
+# For Key_Shared subscriptions, if messages cannot be dispatched to consumers due to a slow consumer
+# or a blocked key hash (because of ordering constraints), the broker will continue reading more
+# messages from the backlog and attempt to dispatch them to consumers until the number of replay
+# messages reaches the calculated threshold.
+# Formula: threshold = max(keySharedLookAheadMsgInReplayThresholdPerConsumer *
+# connected consumer count, keySharedLookAheadMsgInReplayThresholdPerSubscription).
+# Setting this value to 0 will disable the limit calculated per consumer.
+keySharedLookAheadMsgInReplayThresholdPerConsumer=1000
+
+# For Key_Shared subscriptions, if messages cannot be dispatched to consumers due to a slow consumer
+# or a blocked key hash (because of ordering constraints), the broker will continue reading more
+# messages from the backlog and attempt to dispatch them to consumers until the number of replay
+# messages reaches the calculated threshold.
+# Formula: threshold = max(keySharedLookAheadMsgInReplayThresholdPerConsumer *
+# connected consumer count, keySharedLookAheadMsgInReplayThresholdPerSubscription).
+# This value should be set to a value less than 2 * managedLedgerMaxUnackedRangesToPersist.
+# Setting this value to 0 will disable the limit calculated per subscription.
+keySharedLookAheadMsgInReplayThresholdPerSubscription=10000
+
 # Broker periodically checks if subscription is stuck and unblock if flag is enabled. (Default is disabled)
 unblockStuckSubscriptionEnabled=false
 

conf/standalone.conf

@@ -232,6 +232,25 @@ maxUnackedMessagesPerBroker=0
 # limit/2 messages
 maxUnackedMessagesPerSubscriptionOnBrokerBlocked=0.16
 
+# For Key_Shared subscriptions, if messages cannot be dispatched to consumers due to a slow consumer
+# or a blocked key hash (because of ordering constraints), the broker will continue reading more
+# messages from the backlog and attempt to dispatch them to consumers until the number of replay
+# messages reaches the calculated threshold.
+# Formula: threshold = max(keySharedLookAheadMsgInReplayThresholdPerConsumer *
+# connected consumer count, keySharedLookAheadMsgInReplayThresholdPerSubscription).
+# Setting this value to 0 will disable the limit calculated per consumer.
+keySharedLookAheadMsgInReplayThresholdPerConsumer=1000
+
+# For Key_Shared subscriptions, if messages cannot be dispatched to consumers due to a slow consumer
+# or a blocked key hash (because of ordering constraints), the broker will continue reading more
+# messages from the backlog and attempt to dispatch them to consumers until the number of replay
+# messages reaches the calculated threshold.
+# Formula: threshold = max(keySharedLookAheadMsgInReplayThresholdPerConsumer *
+# connected consumer count, keySharedLookAheadMsgInReplayThresholdPerSubscription).
+# This value should be set to a value less than 2 * managedLedgerMaxUnackedRangesToPersist.
+# Setting this value to 0 will disable the limit calculated per subscription.
+keySharedLookAheadMsgInReplayThresholdPerSubscription=10000
+
 # Tick time to schedule task that checks topic publish rate limiting across all topics
 # Reducing to lower value can give more accuracy while throttling publish but
 # it uses more CPU to perform frequent check. (Disable publish throttling with value 0)

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java

@@ -949,6 +949,36 @@ The max allowed delay for delayed delivery (in milliseconds). If the broker rece
             + " back and unack count reaches to `limit/2`. Using a value of 0, is disabling unackedMessage-limit"
             + " check and broker doesn't block dispatchers")
     private int maxUnackedMessagesPerBroker = 0;
+
+    @FieldContext(
+            category = CATEGORY_POLICIES,
+            doc = "For Key_Shared subscriptions, if messages cannot be dispatched to consumers due to a slow consumer"
+                    + " or a blocked key hash (because of ordering constraints), the broker will continue reading more"
+                    + " messages from the backlog and attempt to dispatch them to consumers until the number of replay"
+                    + " messages reaches the calculated threshold.\n"
+                    + "Formula: threshold = max(keySharedLookAheadMsgInReplayThresholdPerConsumer *"
+                    + " connected consumer count, keySharedLookAheadMsgInReplayThresholdPerSubscription)"
+                    + ".\n"
+                    + "Setting this value to 0 will disable the limit calculated per consumer.",
+            dynamic = true
+    )
+    private int keySharedLookAheadMsgInReplayThresholdPerConsumer = 1000;
+
+    @FieldContext(
+            category = CATEGORY_POLICIES,
+            doc = "For Key_Shared subscriptions, if messages cannot be dispatched to consumers due to a slow consumer"
+                    + " or a blocked key hash (because of ordering constraints), the broker will continue reading more"
+                    + " messages from the backlog and attempt to dispatch them to consumers until the number of replay"
+                    + " messages reaches the calculated threshold.\n"
+                    + "Formula: threshold = max(keySharedLookAheadMsgInReplayThresholdPerConsumer *"
+                    + " connected consumer count, keySharedLookAheadMsgInReplayThresholdPerSubscription)"
+                    + ".\n"
+                    + "This value should be set to a value less than 2 * managedLedgerMaxUnackedRangesToPersist.\n"
+                    + "Setting this value to 0 will disable the limit calculated per subscription.\n",
+            dynamic = true
+    )
+    private int keySharedLookAheadMsgInReplayThresholdPerSubscription = 10000;
+
     @FieldContext(
         category = CATEGORY_POLICIES,
         doc = "Once broker reaches maxUnackedMessagesPerBroker limit, it blocks subscriptions which has higher "

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageRedeliveryController.java

@@ -22,7 +22,10 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.NavigableSet;
+import java.util.Optional;
 import java.util.Set;
+import java.util.TreeSet;
+import java.util.function.Predicate;
 import javax.annotation.concurrent.NotThreadSafe;
 import org.apache.bookkeeper.mledger.Position;
 import org.apache.bookkeeper.mledger.PositionFactory;
@@ -146,8 +149,33 @@ public boolean containsStickyKeyHashes(Set<Integer> stickyKeyHashes) {
         return false;
     }
 
-    public NavigableSet<Position> getMessagesToReplayNow(int maxMessagesToRead) {
-        return messagesToRedeliver.items(maxMessagesToRead, PositionFactory::create);
+    public boolean containsStickyKeyHash(int stickyKeyHash) {
+        return !allowOutOfOrderDelivery && hashesRefCount.containsKey(stickyKeyHash);
+    }
+
+    public Optional<Position> getFirstPositionInReplay() {
+        NavigableSet<Position> items = messagesToRedeliver.items(1, PositionFactory::create);
+        return items.isEmpty() ? Optional.empty() : Optional.of(items.first());
+    }
+
+    /**
+     * Get the messages to replay now.
+     *
+     * @param maxMessagesToRead
+     *            the max messages to read
+     * @param filter
+     *            the filter to use to select the messages to replay
+     * @return the messages to replay now
+     */
+    public NavigableSet<Position> getMessagesToReplayNow(int maxMessagesToRead, Predicate<Position> filter) {
+        NavigableSet<Position> items = new TreeSet<>();
+        messagesToRedeliver.processItems(PositionFactory::create, item -> {
+            if (filter.test(item)) {
+                items.add(item);
+            }
+            return items.size() < maxMessagesToRead;
+        });
+        return items;
     }
 
     /**