[pulsar] branch branch-2.7 updated: Fix issue where Key_Shared consumers could get stuck (#10920)

[mm...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

mmerli pushed a commit to branch branch-2.7
in repository https://gitbox.apache.org/repos/asf/pulsar.git


The following commit(s) were added to refs/heads/branch-2.7 by this push:
     new 4cf3569  Fix issue where Key_Shared consumers could get stuck (#10920)
4cf3569 is described below

commit 4cf35698f9d7b0e1977d91a13638985400c9a2b3
Author: Masahiro Sakamoto <ma...@yahoo-corp.jp>
AuthorDate: Tue Jun 15 06:50:59 2021 +0900

    Fix issue where Key_Shared consumers could get stuck (#10920)
---
 .../PersistentStickyKeyDispatcherMultipleConsumers.java           | 8 +++++++-
 1 file changed, 7 insertions(+), 1 deletion(-)

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java
index ca078ac..27bcf97 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java
@@ -123,8 +123,14 @@ public class PersistentStickyKeyDispatcherMultipleConsumers extends PersistentDi
 
     @Override
     public synchronized void removeConsumer(Consumer consumer) throws BrokerServiceException {
-        super.removeConsumer(consumer);
+        // The consumer must be removed from the selector before calling the superclass removeConsumer method.
+        // In the superclass removeConsumer method, the pending acks that the consumer has are added to
+        // messagesToRedeliver. If the consumer has not been removed from the selector at this point,
+        // the broker will try to redeliver the messages to the consumer that has already been closed.
+        // As a result, the messages are not redelivered to any consumer, and the mark-delete position does not move,
+        // eventually causing all consumers to get stuck.
         selector.removeConsumer(consumer);
+        super.removeConsumer(consumer);
         if (recentlyJoinedConsumers != null) {
             recentlyJoinedConsumers.remove(consumer);
             if (consumerList.size() == 1) {