[broker] Fix issue where Key_Shared consumers could get stuck

[massakam]

Motivation

Repeatedly opening and closing consumers on a Key_Shared subscription can occasionally stop dispatching to all consumers. The following is stats of the topic when the phenomenon occurred.

stats.json

{
  "msgRateIn" : 0.0,
  "msgThroughputIn" : 0.0,
  "msgRateOut" : 0.0,
  "msgThroughputOut" : 0.0,
  "averageMsgSize" : 0.0,
  "storageSize" : 888206,
  "backlogSize" : 153264,
  "publishers" : [ ],
  "subscriptions" : {
    "sub1" : {
      "msgRateOut" : 0.0,
      "msgThroughputOut" : 0.0,
      "msgRateRedeliver" : 0.0,
      "msgBacklog" : 2324,
      "blockedSubscriptionOnUnackedMsgs" : false,
      "msgDelayed" : 0,
      "unackedMessages" : 1,
      "type" : "Key_Shared",
      "msgRateExpired" : 0.0,
      "consumers" : [ {
        "msgRateOut" : 0.0,
        "msgThroughputOut" : 0.0,
        "msgRateRedeliver" : 0.0,
        "consumerName" : "5a9ed",
        "availablePermits" : 607,
        "unackedMessages" : 0,
        "blockedConsumerOnUnackedMsgs" : false,
        "metadata" : { },
        "connectedSince" : "2021-06-11T18:57:08.475+09:00",
        "clientVersion" : "2.4.2.36-yjrelease",
        "address" : "/xxx.xxx.xxx.xxx:59196"
      }, {
        "msgRateOut" : 0.0,
        "msgThroughputOut" : 0.0,
        "msgRateRedeliver" : 0.0,
        "consumerName" : "0e74e",
        "availablePermits" : 686,
        "unackedMessages" : 0,
        "blockedConsumerOnUnackedMsgs" : false,
        "metadata" : { },
        "connectedSince" : "2021-06-11T18:57:31.293+09:00",
        "clientVersion" : "2.4.2.36-yjrelease",
        "address" : "/xxx.xxx.xxx.xxx:59198"
      }, {
        "msgRateOut" : 0.0,
        "msgThroughputOut" : 0.0,
        "msgRateRedeliver" : 0.0,
        "consumerName" : "b8ac0",
        "availablePermits" : 952,
        "unackedMessages" : 0,
        "blockedConsumerOnUnackedMsgs" : false,
        "metadata" : { },
        "connectedSince" : "2021-06-11T18:57:58.618+09:00",
        "clientVersion" : "2.4.2.36-yjrelease",
        "address" : "/xxx.xxx.xxx.xxx:59188"
      }, {
        "msgRateOut" : 0.0,
        "msgThroughputOut" : 0.0,
        "msgRateRedeliver" : 0.0,
        "consumerName" : "43e0a",
        "availablePermits" : 1000,
        "unackedMessages" : 0,
        "blockedConsumerOnUnackedMsgs" : false,
        "metadata" : { },
        "connectedSince" : "2021-06-11T18:58:24.501+09:00",
        "clientVersion" : "2.4.2.36-yjrelease",
        "address" : "/xxx.xxx.xxx.xxx:59190"
      } ],
      "isReplicated" : false,
      "consumersAfterMarkDeletePosition" : {
        "43e0a" : "3860483:12549"
      }
    }
  },
  "replication" : { },
  "deduplicationStatus" : "Disabled"
}

The strange thing is that every consumer has an unackedMessages value of 0, but the subscription-level unackedMessages value is 1.

Modifications

The cause of this issue is the following part:

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

Lines 124 to 125 in 894d92b

	super.removeConsumer(consumer);
	selector.removeConsumer(consumer);

When removeConsumer() of the superclass is called, the pending acks owned by that consumer are added to messagesToRedeliver.

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

Lines 184 to 188 in 894d92b

	consumer.getPendingAcks().forEach((ledgerId, entryId, batchSize, none) -> {
	if (addMessageToReplay(ledgerId, entryId)) {
	redeliveryTracker.addIfAbsent(PositionImpl.get(ledgerId, entryId));
	}
	});

However, the consumer has not yet been removed from selector, so the broker attempts to send messages to the consumer that has already been closed. Those messages are removed from messagesToRedeliver, but they aren't actually sent to any consumer.

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

Lines 204 to 210 in 894d92b

	// 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);
	messagesToRedeliver.remove(entry.getLedgerId(), entry.getEntryId());
	}
	}

As a result, the mark-delete position does not move and all consumers will get stuck.

Therefore, in PersistentStickyKeyDispatcherMultipleConsumers#removeConsumer(), we need to remove the consumer from selector before calling removeConsumer() of the superclass.

[codelipenghui]

If the consumer has not been removed from the selector at this point

Looks like a race condition between sending messages to the consumer and remove the consumer from the selector?

We have a synchronized to protect the readEntriesComplete and removeConsumer. How can this happen?

[massakam]

@codelipenghui Perhaps this happens if the messages to be redelivered is in the managed ledger cache. In this case,

readMoreEntries()
↓
readEntriesComplete()
↓
sendMessagesToConsumers()

are executed and completed synchronously in PersistentDispatcherMultipleConsumers#removeConsumer().

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

Lines 184 to 195 in 894d92b

	consumer.getPendingAcks().forEach((ledgerId, entryId, batchSize, none) -> {
	if (addMessageToReplay(ledgerId, entryId)) {
	redeliveryTracker.addIfAbsent(PositionImpl.get(ledgerId, entryId));
	}
	});
	totalAvailablePermits -= consumer.getAvailablePermits();
	if (log.isDebugEnabled()) {
	log.debug("[{}] Decreased totalAvailablePermits by {} in PersistentDispatcherMultipleConsumers. "
	+ "New dispatcher permit count is {}", name, consumer.getAvailablePermits(),
	totalAvailablePermits);
	}
	readMoreEntries();

synchronized protection does not work because all of these methods are executed by the same thread.

[merlimat]

Ok, I think the stack trace here would be like:

removeConsumer()
↓
super.removeConsumer()
↓
readMoreEntries() (This is what is used to trigger the re-delivery of the messages that were pending on the removed consumer)
↓
readEntriesComplete()
↓
sendMessagesToConsumers() As mentioned, this can select the removed consumer which is still in the selector list.

At this point, the sendMessagesToConsumers() will fail and the message will stay into the pendingAcks set for that consumer, but, since the consumer was already removed, the redelivery of this message will not happen.

I think this change is the correct one.

[codelipenghui]

Oh, I see. Thanks for the explanation @massakam @merlimat. I missed that super.removeConsumer() also get a chance to call readMoreEntries().

[merlimat]

👍