[pulsar] branch master updated: [ML] When skipping updating mark delete position, execute callback with executor to prevent deadlock (#15971)

[eo...@apache.org]

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

eolivelli pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git


The following commit(s) were added to refs/heads/master by this push:
     new bbc404bf6c8 [ML] When skipping updating mark delete position, execute callback with executor to prevent deadlock (#15971)
bbc404bf6c8 is described below

commit bbc404bf6c8778e1e788b2b48af60de925256587
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Wed Jun 8 17:48:49 2022 +0300

    [ML] When skipping updating mark delete position, execute callback with executor to prevent deadlock (#15971)
---
 .../java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java  | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
index 10d8d19eb33..605180710db 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
@@ -1896,7 +1896,8 @@ public class ManagedCursorImpl implements ManagedCursor {
                 log.info("Skipping updating mark delete position to {}. The persisted mark delete position {} "
                         + "is later.", mdEntry.newPosition, persistentMarkDeletePosition);
             }
-            mdEntry.triggerComplete();
+            // run with executor to prevent deadlock
+            ledger.getExecutor().executeOrdered(ledger.getName(), safeRun(() -> mdEntry.triggerComplete()));
             return;
         }
 
@@ -1914,7 +1915,8 @@ public class ManagedCursorImpl implements ManagedCursor {
                 log.info("Skipping updating mark delete position to {}. The mark delete position update "
                         + "in progress {} is later.", mdEntry.newPosition, inProgressLatest);
             }
-            mdEntry.triggerComplete();
+            // run with executor to prevent deadlock
+            ledger.getExecutor().executeOrdered(ledger.getName(), safeRun(() -> mdEntry.triggerComplete()));
             return;
         }