Fixed deadlock in key-shared dispatcher (apache#16660)

rkaw92 · Jul 19, 2022 · a5ec98d · a5ec98d
1 parent 5698b08
commit a5ec98d
Showing 1 changed file with 13 additions and 7 deletions.
diff --git a/...ache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java b/...ache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java
@@ -403,13 +403,19 @@ private int getRestrictedMaxEntriesForConsumer(Consumer consumer, List<Entry> en
     }
 
     @Override
-    public synchronized void markDeletePositionMoveForward() {
-        if (recentlyJoinedConsumers != null && !recentlyJoinedConsumers.isEmpty()
-                && removeConsumersFromRecentJoinedConsumers()) {
-            // After we process acks, we need to check whether the mark-delete position was advanced and we can finally
-            // read more messages. It's safe to call readMoreEntries() multiple times.
-            readMoreEntries();
-        }
+    public void markDeletePositionMoveForward() {
+        // Execute the notification in different thread to avoid a mutex chain here
+        // from the delete operation that was completed
+        topic.getBrokerService().getTopicOrderedExecutor().execute(() -> {
+            synchronized (PersistentStickyKeyDispatcherMultipleConsumers.this) {
+                if (recentlyJoinedConsumers != null && !recentlyJoinedConsumers.isEmpty()
+                        && removeConsumersFromRecentJoinedConsumers()) {
+                    // After we process acks, we need to check whether the mark-delete position was advanced and we
+                    // can finally read more messages. It's safe to call readMoreEntries() multiple times.
+                    readMoreEntries();
+                }
+            }
+        });
     }
 
     private boolean removeConsumersFromRecentJoinedConsumers() {