Fix the wrong multi-topic has message available behavior (apache#13634)

Fixes apache#13605 ### Motivation Currently, the multiTopicReader `hasMessageAvailable` might get the wrong result, we must check `numMessagesInQueue() > 0` again after finish all consumer `hasMessageAvaliableAsync` future, bacause some message might already in `MultiTopicsConsumerImpl#incomingMessages`. ### Modifications * Fix the wrong multi-topic has message available behavior. * Use `reader.readNextAsync()` instead of block method `reader.readNext()`. * Reduce the units test running time by changing `MultiTopicsReaderTest` to use `@BeforeClass`, `@AfterClass`.
blitznihar · Jan 12, 2022 · e57dc8f · e57dc8f
1 parent 1d9e7b5
commit e57dc8f
Show file tree

Hide file tree

Showing 2 changed files with 14 additions and 14 deletions.
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MultiTopicsReaderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MultiTopicsReaderTest.java
@@ -61,17 +61,17 @@
 import org.apache.pulsar.common.util.Murmur3_32Hash;
 import org.awaitility.Awaitility;
 import org.testng.Assert;
-import org.testng.annotations.AfterMethod;
-import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
 import org.testng.annotations.Test;
 
-@Test(groups = "flaky")
 @Slf4j
+@Test(groups = "flaky")
 public class MultiTopicsReaderTest extends MockedPulsarServiceBaseTest {
 
     private static final String subscription = "reader-multi-topics-sub";
 
-    @BeforeMethod(alwaysRun = true)
+    @BeforeClass(alwaysRun = true)
     @Override
     protected void setup() throws Exception {
         super.internalSetup();
@@ -87,7 +87,7 @@ protected void setup() throws Exception {
         admin.namespaces().createNamespace("my-property/my-ns", policies);
     }
 
-    @AfterMethod(alwaysRun = true)
+    @AfterClass(alwaysRun = true)
     @Override
     protected void cleanup() throws Exception {
         super.internalCleanup();
@@ -169,15 +169,15 @@ public void testHasMessageAvailableAsync() throws Exception {
     private static <T> void readMessageUseAsync(Reader<T> reader, List<Message<T>> msgs, CountDownLatch latch) {
         reader.hasMessageAvailableAsync().thenAccept(hasMessageAvailable -> {
             if (hasMessageAvailable) {
-                try {
-                    Message<T> msg = reader.readNext();
+                reader.readNextAsync().whenComplete((msg, ex) -> {
+                    if (ex != null) {
+                        log.error("Read message failed.", ex);
+                        latch.countDown();
+                        return;
+                    }
                     msgs.add(msg);
-                } catch (PulsarClientException e) {
-                    log.error("Read message failed.", e);
-                    latch.countDown();
-                    return;
-                }
-                readMessageUseAsync(reader, msgs, latch);
+                    readMessageUseAsync(reader, msgs, latch);
+                });
             } else {
                 latch.countDown();
             }

diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java
@@ -782,7 +782,7 @@ public CompletableFuture<Boolean> hasMessageAvailableAsync() {
             if (exception != null) {
                 completableFuture.completeExceptionally(exception);
             } else {
-                completableFuture.complete(hasMessageAvailable.get());
+                completableFuture.complete(hasMessageAvailable.get() || numMessagesInQueue() > 0);
             }
         });
         return completableFuture;