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

[lhotari]

Fixes #23200

Motivation

This PR introduces enhancements to the Key_Shared subscription dispatching logic in the Pulsar broker. The changes aim to optimize message dispatching, improve the handling of slow consumers, and address several identified issues related to message replay and message "look ahead" when consumers are blocked in delivery due to key ordering restrictions.

This PR adds limits to the Key_Shared Subscription look-ahead feature, which was introduced in PR #7105. Additionally, there are several improvements to the previous logic.

Since there wasn't any named concept for the feature added in PR #7105, it has been named "key shared look ahead" in this PR. This references the feature added in PR #7105, where the dispatcher will skip replaying messages and read more messages from the backlog. The changes in PR #7105 didn't add a limit for reading more messages. The comments in the PR mention that it will be limited by the unacked message limits. However, this isn't the case since the read messages aren't considered unacked messages until they have been dispatched to consumers.

Reading ahead would be triggered in multiple cases:

• When consumers are backpressured due to slow processing or message delivery
• When consumers are blocked due to key ordering restrictions
  • The previous logic for blocking a hash was unoptimal. The impact is discussed in a Pulsar Q&A discussion. It seemed that a hash range got blocked, but that wasn't the case.
    • The replay queue had a head-of-line blocking issue, which caused many hashes being blocked unnecessarily
    • In reading more entries, all entries got dropped when any of the read entries was blocked by a hash.
      • This is unnecessary for retaining ordering. This is addressed in this PR.
        • It seems that addressing this issues depends on PIP-282 changes, [fix][broker] Introduce the last sent position to fix message ordering issues in Key_Shared (PIP-282) #21953.

Since consumers were blocked unnecessarily, reading ahead would often just continue reading the complete backlog into the replay queue. The impact is described in #23200

• Reading the entries to the replay queue causes unnecessary network traffic, disk IO & CPU on bookies and CPU on brokers. The read entries will be dropped in most cases.
  • Issue [improve] [broker] Avoid repeated Read-and-discard when using Key_Shared mode #22245 has been a previous attempt to mitigate this symptom
• Reading ahead would cause a lot of "ack holes".
  • PIP-299 [improve] [pip] PIP-299: Stop dispatch messages if the individual acks will be lost in the persistent storage #21118 has been a previous attempt to mitigate this symptom

In addition to fix the infinite key shared look ahead issue, this PR seems to fix a problem in the existing implementation where this code blocks progress of a consumer:

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

Lines 453 to 458 in d7e8ea1

	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;
	}

What this means is that when the dispatcher has read a batch, it will discard all messages in the read messages if any hash is present in the replay queue (redelivery messages). This is an unoptimal solution for ensuring that ordering is preserved. This PR addresses the issue while attempting to keep backwards compatibility from a single consumer's perspective.

The "look ahead" limit introduced in this PR will add backpressure to how the dispatcher works. Eventually consumers will be slower than the dispatcher sending messages to consumers and that's why it's necessary to have the limit, also in cases where the hashes wouldn't be blocked or "allowOutOfOrderDelivery" mode is used.

There's also another related problem that this change addresses. The head of the replay queue might get blocked in the release Pulsar versions.

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

Lines 348 to 350 in ed14f21

	NavigableSet<Position> messagesToReplayNow = getMessagesToReplayNow(messagesToRead);
	NavigableSet<Position> messagesToReplayFiltered = filterOutEntriesWillBeDiscarded(messagesToReplayNow);
	if (!messagesToReplayFiltered.isEmpty()) {

In this PR, the solution to this is to take as many entries that match the criteria from the replay queue. This required changing the getMessagesToReplayNow to accept a Predicate for filtering:

    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;
    }

Modifications

1. Configuration Properties:

   • Added new configuration properties to control the limits of Key_Shared subscriptions look ahead in dispatching:
     • keySharedLookAheadMsgInReplayThresholdPerConsumer
     • keySharedLookAheadMsgInReplayThresholdPerSubscription
     • keySharedLookAheadEnabledWhenRecentlyJoinedConsumersExist

2. Broker Configuration:

   • Updated broker.conf to include descriptions and default values for the new configuration properties.

3. Key_Shared Subscription Logic:

   • Implement the key shared look-ahead limits that are configured with the above configuration properties.
   • Enhance the dispatching to ensure efficient dispatching:
     • Instead of filtering the list of entries pulled from the message redelivery controller, refactor the solution to take a function that can filter the entries until there's a sufficient amount of messages for dispatching.
       • This refactoring eliminates the previous method filterOutEntriesWillBeDiscarded while preserving the previous logic of filtering.
     • Remove the unnecessary thread-local fields localGroupedEntries and the unused localGroupedPositions.
       • It is unnecessary to eliminate the creation of short-lived objects in modern JVMs. It just clutters the code without any performance benefit. "No garbage" style is relevant mainly for long-lived objects stored in caches or large array allocations.
   • Replace getMessagesToReplayNow(1) with getFirstPositionInReplay().
     • This check will no longer have the side effect of pulling messages from delayed delivery when it's called.
   • Add a new method canReplayMessages to make the look-ahead logic more explicit. The previous solution of returning an empty set from the getMessagesToReplayNow method made it hard to understand how the look-ahead behavior is implemented.
   • Rename the isDispatcherStuckOnReplays field to skipNextReplayToTriggerLookAhead to make the meaning of the field explicit and self-descriptive.
   • Fix issue where the !havePendingReplayRead status wasn't checked before starting a new read.
     • For shared subscription, it's fine to just check !havePendingPendingRead, but for Key_Shared it isn't.
     • extract doesntHavePendingRead for handling the logic
   • Rename the keyNumbers variable to remainingConsumersToFinishSending so that the meaning of the variable is explicit and self-descriptive.
   • Rename the groupedEntries variable to entriesByConsumerForDispatching so that the meaning of the variable is explicit and self-descriptive.
   • Rename the entriesWithSameKey variable to entriesForConsumer so that the variable is self-descriptive.
   • Fix calculations based on maxUnackedMessages and unackedMessages. The unit is in individual messages and permits are in entries. These 2 cannot be mixed.
   • Addressing the consumer blocked issue: the logic has been changed to evaluate each message hash one-by-one.

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

[lhotari]

This PR is ready for initial review feedback. I'm looking into ways of adding tests for the key_shared look ahead limits.

Outdated comment

[poorbarcode]

@lhotari

• Three weeks ago: I submitted a Request Change

• Two weeks ago, you left a comment: Noticing more challenges to cover while getting into the tests. Setting this to draft mode until the concerns are addressed and tests have been added. Some of the concerns are related to [Bug] Consumers getting stuck since read more entries doesn't properly handle retries with backoff #23264.

• 5 days ago: you dismissed the Request Change, and left a comment: "Outdated comment"

Between step-2 and step-3, you never marked it as available to review and never ask to review again, then you merged the PR 😂

[lhotari]

I'm sorry @poorbarcode. Please let me explain the misunderstanding.

5 days ago: you dismissed the Request Change, and left a comment: "Outdated comment"

For me, the "request for changes" comment was "Please do not merge this PR fast, I will finish my review tomorrow" (#23231 (review)). That's why I dismissed it with "Outdated comment". I had already addressed also the other review comments you had made.

Between step-2 and step-3, you never marked it as available to review and never ask to review again, then you merged the PR 😂

I'm sorry for that. I did request a review, but you didn't respond. You can check the timeline. The PR was in draft mode for some time, but the reason for that was that the PR was already approved and I didn't want anyone to merge it. I didn't realize that you wouldn't be reviewing it because of the draft status.

The changes in this PR aren't final, if you have review feedback, it's possible also address that after the PR has been merged. Please add comments to this PR of any concerns you may have.

[poorbarcode]

@lhotari

Sure, thanks for replying this comment. ❤️