[fix][broker] Multiple consumer dispatcher stuck when unackedMessages greater than maxUnackedMessages

[mattisonchao]

Motivation

See error stack trace:

[Unreachable] char[1024] “2022-08-31T08:05:02,202+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] ERROR org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl - Unknown exception for ManagedLedgerException.
java.lang.ArrayIndexOutOfBoundsException: -22012
	at java.util.ArrayList.elementData(ArrayList.java:424) ~[?:1.8.0_322]
	at java.util.ArrayList.get(ArrayList.java:437) ~[?:1.8.0_322]
	at org.apache.pulsar.broker.service.persistent.PersistentStickyKeyDispatcherMultipleConsumers.sendMessagesToConsumers(PersistentStickyKeyDispatcherMultipleConsumers.java:256)
	at org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers.readEntriesComplete(PersistentDispatcherMultipleConsumers.java:517)
	at org.apache.bookkeeper.mledger.impl.ManagedCursorImpl$12.readEntryComplete(ManagedCursorImpl.java:1392)
	at org.apache.bookkeeper.mledger.impl.EntryCacheImpl.”

Relative code:

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

Lines 237 to 262 in cfe95dd

	int availablePermits = consumer == null ? 0 : Math.max(consumer.getAvailablePermits(), 0);
	if (consumer != null && consumer.getMaxUnackedMessages() > 0) {
	availablePermits = Math.min(availablePermits,
	consumer.getMaxUnackedMessages() - consumer.getUnackedMessages());
	}
	int maxMessagesForC = Math.min(entriesWithSameKeyCount, availablePermits);
	int messagesForC = getRestrictedMaxEntriesForConsumer(consumer, entriesWithSameKey, maxMessagesForC,
	readType, consumerStickyKeyHashesMap.get(consumer));
	if (log.isDebugEnabled()) {
	log.debug("[{}] select consumer {} with messages num {}, read type is {}",
	name, consumer == null ? "null" : 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) {

When unackedMessages is greater than maxUnackedMessages, because messagesForC is a negative number, ArrayIndexOutOfBoundsException will be thrown on line 254. Then the exception will cause the message not to be delivered.

Modifications

• Add check to avoid negative results

Verifying this change

• Make sure that the change passes the CI checks.

Documentation

• doc-not-needed
  (Please explain why)

[lhotari]

at org.apache.pulsar.broker.service.persistent.PersistentStickyKeyDispatcherMultipleConsumers.sendMessagesToConsumers(PersistentStickyKeyDispatcherMultipleConsumers.java:256)

which Pulsar version is this stack trace from?

Would it be possible to add a test that reproduces the problem using the public APIs? If this happens when there are more unknowledged messages than the defined limit, I'd assume that it's fairly easy to reproduce?

[codelipenghui]

@mattisonchao I think the issue is introduced by #16718? it should be a blocker for 2.11.0 release.

[mattisonchao]

@lhotari Test added, Please take a look.

[michaeljmarshall]

The solution makes sense to me, but I'm wondering if there is another bug that we need to fix too? Thanks!

[michaeljmarshall]

Do we understand why c.getUnackedMessages() is greater than c.getMaxUnackedMessages()? I read through the test, but I didn't see what is causing the negative availablePermits.

Based on the names, that seems like there is another bug if c.getMaxUnackedMessages() - c.getUnackedMessages() is negative. Perhaps the idea that the maxUnackedMessages is just a soft limit that we shouldn't exceed by too much?

[mattisonchao]

Do we understand why c.getUnackedMessages() is greater than c.getMaxUnackedMessages()?

Because in the current implementation, we will add unackedMessages after sent messages. if we use batch messages, the unackedMessages is probably greater than MaxUnackedMessages.

Plus, #16670 #16718 want to add this check to limit the consumer.

Based on the names, that seems like there is another bug if c.getMaxUnackedMessages() - c.getUnackedMessages() is negative.

Sure, But I find we have two sections using it. And I fix It all in this PR.

Perhaps the idea that the maxUnackedMessages is just a soft limit that we shouldn't exceed by too much?

Yes, maybe we have another method to help more robust support it, but it looks like it is another improvement.

[michaeljmarshall]

Because in the current implementation, we will add unackedMessages after sent messages. if we use batch messages, the unackedMessages is probably greater than MaxUnackedMessages.

This was the detail I was missing, thank you for explaining. I read through the code a bit more, and it makes sense.

[congbobo184]

If not carry this fix in master, the test for share sub mode can passed

[eolivelli]

LGTM