[fix] [broker] Subscription stuck due to called Admin API analyzeSubscriptionBacklog

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java

@@ -795,7 +795,8 @@ public void asyncReadEntriesWithSkip(int numberOfEntriesToRead, long maxSizeByte
         // Skip deleted entries.
         skipCondition = skipCondition == null ? this::isMessageDeleted : skipCondition.or(this::isMessageDeleted);
         OpReadEntry op =
-                OpReadEntry.create(this, readPosition, numOfEntriesToRead, callback, ctx, maxPosition, skipCondition);
+                OpReadEntry.create(this, readPosition, numOfEntriesToRead, callback, ctx, maxPosition,
+                        skipCondition, true);
         ledger.asyncReadEntries(op);
     }
 
@@ -955,7 +956,7 @@ public void asyncReadEntriesWithSkipOrWait(int maxEntries, long maxSizeBytes, Re
             // Skip deleted entries.
             skipCondition = skipCondition == null ? this::isMessageDeleted : skipCondition.or(this::isMessageDeleted);
             OpReadEntry op = OpReadEntry.create(this, readPosition, numberOfEntriesToRead, callback,
-                    ctx, maxPosition, skipCondition);
+                    ctx, maxPosition, skipCondition, true);
 
             if (!WAITING_READ_OP_UPDATER.compareAndSet(this, null, op)) {
                 op.recycle();

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpReadEntry.java

@@ -48,8 +48,11 @@ class OpReadEntry implements ReadEntriesCallback {
 
     Predicate<PositionImpl> skipCondition;
 
+    private boolean moveCursorReadPos;
+
     public static OpReadEntry create(ManagedCursorImpl cursor, PositionImpl readPositionRef, int count,
-            ReadEntriesCallback callback, Object ctx, PositionImpl maxPosition, Predicate<PositionImpl> skipCondition) {
+            ReadEntriesCallback callback, Object ctx, PositionImpl maxPosition, Predicate<PositionImpl> skipCondition,
+                                     boolean moveCursorReadPos) {
         OpReadEntry op = RECYCLER.get();
         op.readPosition = cursor.ledger.startReadOperationOnLedger(readPositionRef);
         op.cursor = cursor;
@@ -63,6 +66,7 @@ public static OpReadEntry create(ManagedCursorImpl cursor, PositionImpl readPosi
         op.skipCondition = skipCondition;
         op.ctx = ctx;
         op.nextReadPosition = PositionImpl.get(op.readPosition);
+        op.moveCursorReadPos = moveCursorReadPos;
         return op;
     }
 
@@ -92,7 +96,9 @@ void internalReadEntriesComplete(List<Entry> returnedEntries, Object ctx, Positi
         // if entries have been filtered out then try to skip reading of already deletedMessages in that range
         final Position nexReadPosition = entriesCount != filteredEntries.size()
                 ? cursor.getNextAvailablePosition(lastPosition) : lastPosition.getNext();
-        updateReadPosition(nexReadPosition);
+        if (moveCursorReadPos) {
+            updateReadPosition(nexReadPosition);
+        }
         checkReadCompletion();
     }
 
@@ -161,9 +167,7 @@ void updateReadPosition(Position newReadPosition) {
 
     void checkReadCompletion() {
         // op readPosition is smaller or equals maxPosition then can read again
-        if (entries.size() < count && cursor.hasMoreEntries()
-                && maxPosition.compareTo(readPosition) > 0) {
-
+        if (needAccumulateMoreEntries()) {
             // We still have more entries to read from the next ledger, schedule a new async operation
             cursor.ledger.getExecutor().execute(() -> {
                 readPosition = cursor.ledger.startReadOperationOnLedger(nextReadPosition);
@@ -183,6 +187,16 @@ void checkReadCompletion() {
         }
     }
 
+    private boolean needAccumulateMoreEntries() {
+        if (!moveCursorReadPos) {
+            // Just responds the entries of current reads, because "cursor.readPosition" has not moved, so cannot
+            // accumulate more entries.
+            return false;
+        }
+        boolean hasMoreEntries = cursor.hasMoreEntries() && maxPosition.compareTo(readPosition) > 0;
+        return entries.size() < count && hasMoreEntries;
+    }
+
     public int getNumberOfEntriesToRead() {
         return count - entries.size();
     }

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpScan.java

@@ -128,7 +128,7 @@ public void find() {
         }
         if (cursor.hasMoreEntries(searchPosition)) {
             OpReadEntry opReadEntry = OpReadEntry.create(cursor, searchPosition, batchSize,
-            this, OpScan.this.ctx, null, null);
+            this, OpScan.this.ctx, null, null, false);
             ledger.asyncReadEntries(opReadEntry);
         } else {
             callback.scanComplete(lastSeenPosition, ScanOutcome.COMPLETED, OpScan.this.ctx);

managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java

@@ -18,6 +18,7 @@
  */
 package org.apache.bookkeeper.mledger.impl;
 
+import static org.mockito.ArgumentMatchers.anyBoolean;
 import static org.mockito.ArgumentMatchers.anyInt;
 import static org.mockito.Mockito.any;
 import static org.mockito.Mockito.doAnswer;
@@ -2117,7 +2118,7 @@
         assertEquals(ScanOutcome.COMPLETED, c1.scan(Optional.empty(), (entry -> {
             positionsAfterDelete.add(entry.getPosition());
             return true;
         }), batchSize, Long.MAX_VALUE, Long.MAX_VALUE).get());
         assertEquals(numEntries - 1, positionsAfterDelete.size());
 
         // delete all the entries
@@ -4249,7 +4250,7 @@
 
         // op readPosition is bigger than maxReadPosition
         OpReadEntry opReadEntry = OpReadEntry.create(cursor, ledger.lastConfirmedEntry, 10, callback,
-                null, PositionImpl.get(lastPosition.getLedgerId(), -1), null);
+                null, PositionImpl.get(lastPosition.getLedgerId(), -1), null, true);
         Field field = ManagedCursorImpl.class.getDeclaredField("readPosition");
         field.setAccessible(true);
         field.set(cursor, PositionImpl.EARLIEST);
@@ -4271,7 +4272,8 @@
         };
 
         @Cleanup final MockedStatic<OpReadEntry> mockedStaticOpReadEntry = Mockito.mockStatic(OpReadEntry.class);
-        mockedStaticOpReadEntry.when(() -> OpReadEntry.create(any(), any(), anyInt(), any(), any(), any(), any()))
+        mockedStaticOpReadEntry.when(() -> OpReadEntry.create(any(), any(), anyInt(), any(), any(), any(), any(),
+                        anyBoolean()))
                 .thenAnswer(__ -> createOpReadEntry.get());
 
         final ManagedLedgerConfig ledgerConfig = new ManagedLedgerConfig();

managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java

@@ -682,7 +682,7 @@ public void readEntriesComplete(List<Entry> entries, Object ctx) {
                     public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
 
                     }
-                }, null, maxPosition, null);
+                }, null, maxPosition, null, true);
         Assert.assertEquals(opReadEntry.readPosition, position);
     }
 
@@ -3157,7 +3157,7 @@ public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
                 responseException2.set(exception);
             }
 
-        }, null, PositionImpl.LATEST, null);
+        }, null, PositionImpl.LATEST, null, true);
         ledger.asyncReadEntry(ledgerHandle, PositionImpl.EARLIEST.getEntryId(), PositionImpl.EARLIEST.getEntryId(),
                 opReadEntry, ctxStr);
         retryStrategically((test) -> {

pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java

@@ -3389,4 +3389,33 @@ private void testSetBacklogQuotasNamespaceLevelIfRetentionExists() throws Except
         // cleanup.
         admin.namespaces().deleteNamespace(ns);
     }
+
+    @Test
+    private void testAnalyzeSubscriptionBacklogNotCauseStuck() throws Exception {
+        final String topic = BrokerTestUtil.newUniqueName("persistent://" + defaultNamespace + "/tp");
+        final String subscription = "s1";
+        admin.topics().createNonPartitionedTopic(topic);
+        // Send 10 messages.
+        Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING).topic(topic).subscriptionName(subscription)
+                .receiverQueueSize(0).subscribe();
+        Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(topic).create();
+        for (int i = 0; i < 10; i++) {
+            producer.send(i + "");
+        }
+
+        // Verify consumer can receive all messages after calling "analyzeSubscriptionBacklog".
+        admin.topics().analyzeSubscriptionBacklog(topic, subscription, Optional.of(MessageIdImpl.earliest));
+        for (int i = 0; i < 10; i++) {
+            Awaitility.await().untilAsserted(() -> {
+                Message m = consumer.receive();
+                assertNotNull(m);
+                consumer.acknowledge(m);
+            });
+        }
+
+        // cleanup.
+        consumer.close();
+        producer.close();
+        admin.topics().delete(topic);
+    }
 }