fix bug: ignoring flow control message since cursor reset in progress

apache · Jul 12, 2022 · 5cc8392 · 5cc8392
1 parent 3752a11
commit 5cc8392
Show file tree

Hide file tree

Showing 2 changed files with 14 additions and 0 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
@@ -3283,4 +3283,12 @@ public void setState(State state) {
     public ManagedLedgerConfig getConfig() {
         return config;
     }
+
+    /**
+     * check cursor reset status
+     * @return true if the cursor reset in progress
+     */
+    public boolean resetCursorInProgress() {
+        return RESET_CURSOR_IN_PROGRESS_UPDATER.get(this) == TRUE;
+    }
 }
diff --git a/...org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java b/...org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java
@@ -34,6 +34,7 @@
 import org.apache.bookkeeper.mledger.ManagedLedgerException;
 import org.apache.bookkeeper.mledger.ManagedLedgerException.NoMoreEntriesToReadException;
 import org.apache.bookkeeper.mledger.ManagedLedgerException.TooManyRequestsException;
+import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
 import org.apache.bookkeeper.mledger.impl.PositionImpl;
 import org.apache.bookkeeper.mledger.util.SafeRun;
 import org.apache.commons.lang3.tuple.Pair;
@@ -278,6 +279,11 @@ private synchronized void internalConsumerFlow(Consumer consumer) {
                 log.debug("[{}-{}] Ignoring flow control message since consumer is waiting for cursor to be rewinded",
                         name, consumer);
             }
+        } else if (((ManagedCursorImpl) cursor).resetCursorInProgress()) {
+            if (log.isDebugEnabled()) {
+                log.debug("[{}-{}] Ignoring flow control message since cursor reset in progress - cursor {}",
+                        name, consumer, cursor.getName());
+            }
         } else {
             if (log.isDebugEnabled()) {
                 log.debug("[{}-{}] Trigger new read after receiving flow control message", name, consumer);