(pulsar) branch master updated: [fix][test] Fix flaky testPrepareInitPoliciesCacheAsyncThrowExceptionAfterCreateReader (#26049)

[mm...@apache.org]

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

merlimat 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 6f82106ea6c [fix][test] Fix flaky testPrepareInitPoliciesCacheAsyncThrowExceptionAfterCreateReader (#26049)
6f82106ea6c is described below

commit 6f82106ea6c4d508a883ff27875ef279c7c4101c
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Thu Jun 18 02:44:55 2026 +0300

    [fix][test] Fix flaky testPrepareInitPoliciesCacheAsyncThrowExceptionAfterCreateReader (#26049)
---
 .../SystemTopicBasedTopicPoliciesServiceTest.java  | 28 +++++++++++++++++-----
 1 file changed, 22 insertions(+), 6 deletions(-)

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesServiceTest.java
index 2523d1bd49c..b6dbffede81 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesServiceTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesServiceTest.java
@@ -49,6 +49,7 @@ import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
 import org.apache.pulsar.broker.systopic.SystemTopicClient;
 import org.apache.pulsar.client.admin.PulsarAdminException;
 import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.api.Schema;
 import org.apache.pulsar.common.events.PulsarEvent;
 import org.apache.pulsar.common.naming.NamespaceName;
@@ -554,13 +555,28 @@ public class SystemTopicBasedTopicPoliciesServiceTest extends MockedPulsarServic
             assertTrue(logFound);
         });
 
-        // Since cleanPoliciesCacheInitMap() is executed, should add the failed reader into readerCache again.
-        // Then in SystemTopicBasedTopicPoliciesService, readerCache has a closed reader,
-        // and policyCacheInitMap do not contain a future.
-        // To simulate the situation: when getTopicPolicy() execute, it will do prepareInitPoliciesCacheAsync() and
-        // use a closed reader to read the __change_event topic. Then throw exception
-        spyReaderCaches.put(NamespaceName.get(NAMESPACE5), readerCompletableFuture);
+        // The reader.close() above drives readMorePoliciesAsync() into cleanPoliciesCacheInitMap(), which logs
+        // "Closing the topic policies reader for" and removes the namespace from policyCacheInitMap and readerCaches.
+        // Now exercise a follow-up prepareInitPoliciesCacheAsync() that reuses a closed reader and must fail in
+        // initPolicesCache(). Two timing hazards made this flaky (#25081), so both are pinned deterministically:
+        //  1) A real closed reader's hasMoreEventsAsync() (Reader.hasMessageAvailableAsync()) can answer from cached
+        //     state instead of failing with AlreadyClosedException, in which case initPolicesCache() reaches the end
+        //     of the topic and completes successfully and the expected exception never happens. Use a spy whose
+        //     hasMoreEventsAsync() always fails with AlreadyClosedException.
+        //  2) A background topic load can re-run prepareInitPoliciesCacheAsync() for the namespace and leave a
+        //     completed init future behind; the next call would then short-circuit through the existing-future
+        //     branch and never re-initialize. Drop any init future right before the call so it re-initializes, and
+        //     stub createSystemTopicClient() so every reader created for this namespace is the failing spy — then
+        //     whichever initialization wins the race fails in the same (asserted) way.
+        SystemTopicClient.Reader<PulsarEvent> closedReader = Mockito.spy(reader);
+        Mockito.doReturn(CompletableFuture.failedFuture(
+                        new PulsarClientException.AlreadyClosedException("Reader is already closed")))
+                .when(closedReader).hasMoreEventsAsync();
+        Mockito.doReturn(CompletableFuture.completedFuture(closedReader))
+                .when(spyService).createSystemTopicClient(NamespaceName.get(NAMESPACE5));
+        spyReaderCaches.put(NamespaceName.get(NAMESPACE5), CompletableFuture.completedFuture(closedReader));
         FieldUtils.writeDeclaredField(spyService, "readerCaches", spyReaderCaches, true);
+        spyService.policyCacheInitMap.remove(NamespaceName.get(NAMESPACE5));
 
         CompletableFuture<Boolean> prepareFuture = new CompletableFuture<>();
         try {