[broker] Fix issue that message ordering could be broken when redelivering messages on Key_Shared subscription

[massakam]

Motivation

Messages with the same key can be out of order if message redelivery occurs on a Key_Shared subscription.

1. Suppose PersistentDispatcherMultipleConsumers#messagesToRedeliver contains message-1 and message-2. Message-1 will be delivered to consumer-a and message-2 will be delivered to consumer-b.
2. The dispatcher tried to send message-1 to consumer-a, but the consumer was too slow to send it.
3. Consumer-a is added to stuckConsumers.
   

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

   Lines 263 to 266 in 894d92b

   	if (!nextStuckConsumers.isEmpty()) {
   	isDispatcherStuckOnReplays = true;
   	stuckConsumers.addAll(nextStuckConsumers);
   	}

4. The next time readMoreEntries() is run, getMessagesToReplayNow() will return an empty Set because isDispatcherStuckOnReplays is true.
   

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

   Lines 368 to 374 in 894d92b

   	protected synchronized Set<PositionImpl> 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.emptySet();

5. The dispatcher reads newer messages instead of the messages contained in messagesToRedeliver.
   

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

   Lines 233 to 267 in 894d92b

   	Set<PositionImpl> messagesToReplayNow = getMessagesToReplayNow(messagesToRead);
   	if (!messagesToReplayNow.isEmpty()) {
   	if (log.isDebugEnabled()) {
   	log.debug("[{}] Schedule replay of {} messages for {} consumers", name, messagesToReplayNow.size(),
   	consumerList.size());
   	}
   	havePendingReplayRead = true;
   	Set<? extends Position> deletedMessages = topic.isDelayedDeliveryEnabled()
   	? asyncReplayEntriesInOrder(messagesToReplayNow) : asyncReplayEntries(messagesToReplayNow);
   	// clear already acked positions from replay bucket
   	deletedMessages.forEach(position -> messagesToRedeliver.remove(((PositionImpl) position).getLedgerId(),
   	((PositionImpl) position).getEntryId()));
   	// if all the entries are acked-entries and cleared up from messagesToRedeliver, try to read
   	// next entries as readCompletedEntries-callback was never called
   	if ((messagesToReplayNow.size() - deletedMessages.size()) == 0) {
   	havePendingReplayRead = false;
   	// We should not call readMoreEntries() recursively in the same thread
   	// as there is a risk of StackOverflowError
   	topic.getBrokerService().executor().execute(() -> readMoreEntries());
   	}
   	} else if (BLOCKED_DISPATCHER_ON_UNACKMSG_UPDATER.get(this) == TRUE) {
   	log.warn("[{}] Dispatcher read is blocked due to unackMessages {} reached to max {}", name,
   	totalUnackedMessages, topic.getMaxUnackedMessagesOnSubscription());
   	} else if (!havePendingRead) {
   	if (log.isDebugEnabled()) {
   	log.debug("[{}] Schedule read of {} messages for {} consumers", name, messagesToRead,
   	consumerList.size());
   	}
   	havePendingRead = true;
   	cursor.asyncReadEntriesOrWait(messagesToRead, serviceConfig.getDispatcherMaxReadSizeBytes(),
   	this,
   	ReadType.Normal, topic.getMaxReadPosition());

6. A new message (message-3) is delivered to consumer-b.
7. Message-2 contained in messagesToRedeliver is delivered to consumer-b.
8. As a result, the order of message-2 and message-3 is reversed.

Modifications

When adding a message to be redeliver to messagesToRedeliver, save the hash of the key that the message has. If the dispatcher attempts to send newer messages to the consumer that have a key corresponding to any one of the saved hash values, they will be added to messagesToRedeliver instead of being sent. This prevents messages with the same key from being out of order.

[massakam]

This method is executed when redelivery is requested from the consumer side (e.g. negative ack, ack timeout). So the user should allow the messages to be out of order in this case.

[nkurihar]

LGTM

[eolivelli]

great work!

I left one question for curiosity

[eolivelli]

+1

[massakam]

@lhotari PTAL

[eolivelli]

LGTM

@massakam nice idea!

[lhotari]

LGTM

Outstanding work @massakam !

[merlimat]

Using key() will make a copy of the whole set, so we should instead do the forEach().

[massakam]

@merlimat If we try to remove any entry in forEach(), it seems to cause a deadlock. Is there any way to avoid this without using keys()?

[lhotari]

@massakam Does the deadlock happen in 2.8.x / master branch? There are changes in #9787 that resolved some dead lock issues. (see #10691 for more details)

[massakam]

@lhotari It happens in the master branch. hashesToBeBlocked is an instance of org.apache.bookkeeper.util.collections.ConcurrentLongLongPairHashMap, so it should be unrelated to those fixes.

[merlimat]

Ok, we can go on with this but we should also import here a fixed version of ConcurrentLongLongPairHashMap that doesn't have the problem

[massakam]

but we should also import here a fixed version of ConcurrentLongLongPairHashMap that doesn't have the problem

@merlimat Such a version doesn't exist yet, does it?

[massakam]

Add the keys to be removed in the forEach() loop to the newly created list and remove them from hashesToBeBlocked after exiting the loop. This makes the entries to be copied part of the set rather than the whole set. PTAL.
266ca5d

[codelipenghui]

LGTM

[codelipenghui]

@merlimat Please help review this PR again.