[fix][test] Fix flaky PersistentStickyKeyDispatcherMultipleConsumersClassicTest.testSkipRedeliverTemporally (#25385)

lhotari · claude · web-flow · commit 953d092b10ab · 2026-03-22T20:32:33.000-07:00
Co-authored-by: Claude Opus 4.6 (1M context) &lt;noreply@anthropic.com&gt;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersClassicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersClassicTest.java
@@ -499,13 +499,23 @@ public void testSkipRedeliverTemporally() throws InterruptedException {
             return Collections.emptySet();
         }).when(cursorMock).asyncReplayEntries(anySet(), any(), any(), anyBoolean());
 
+        // Simulate real cursor behavior: track read position so entries are only returned once
+        // by normal reads (subsequent access must go through asyncReplayEntries).
+        // When no new entries are available, don't call the callback (simulating "OrWait" behavior).
+        Set<Position> normalReadReturned = new ConcurrentSkipListSet<>();
         doAnswer(invocationOnMock -> {
             int maxEntries = invocationOnMock.getArgument(0);
             AsyncCallbacks.ReadEntriesCallback callback = invocationOnMock.getArgument(2);
             List<Entry> entries = allEntries.stream()
-                    .filter(entry -> entry.getLedgerId() != -1 && !alreadySent.contains(entry.getPosition()))
+                    .filter(entry -> entry.getLedgerId() != -1
+                            && !normalReadReturned.contains(entry.getPosition()))
                     .limit(maxEntries)
                     .toList();
+            if (entries.isEmpty()) {
+                // No new entries available - simulate "wait" by not calling callback
+                return null;
+            }
+            entries.forEach(e -> normalReadReturned.add(e.getPosition()));
             Object ctx = invocationOnMock.getArgument(3);
             callback.readEntriesComplete(copyEntries(entries), ctx);
             return null;
@@ -551,6 +561,11 @@ public void testSkipRedeliverTemporally() throws InterruptedException {
         // set permits to 2
         slowConsumerAvailablePermits.set(2);
 
+        // Trigger a new read cycle so the dispatcher can do a replay read to deliver
+        // messages to the slow consumer. In production, this would be triggered by
+        // consumerFlow when the consumer sends more permits.
+        persistentDispatcher.readMoreEntriesAsync();
+
         // now wait for slow consumer messages since there are permits
         assertTrue(slowConsumerMessagesSent.await(5, TimeUnit.SECONDS));
 
