[pulsar] branch master updated: [fix] [client] fix reader.hasMessageAvailable return false when incoming queue is not empty (#21259)

[yu...@apache.org]

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

yubiao 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 6d82b09128f [fix] [client] fix reader.hasMessageAvailable return false when incoming queue is not empty (#21259)
6d82b09128f is described below

commit 6d82b09128f46fdcb27021560d773fac15d66a48
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Thu Sep 28 23:45:39 2023 +0800

    [fix] [client] fix reader.hasMessageAvailable return false when incoming queue is not empty (#21259)
    
    ### Motivation
    
    Reproduce steps:
    - Create a reader.
    - Reader pulls messages into `incoming queue`, do not call `reader.readNext` now.
    - Trim ledger task will delete the ledgers, then there is no in the topic.
    - Now, you can get messages if you call `reader.readNext`, but the method `reader.hasMessageAvailable` return `false`
    
    Note: the similar issue of `MultiTopicsConsumerImpl` has been fixed by https://github.com/apache/pulsar/pull/13332, current PR only trying to fix the issue of `ConsumerImpl`.
    
    ### Modifications
    
    Make `reader.hasMessageAvailable` return `true` when `incoming queue` is not empty.
---
 .../client/api/NonDurableSubscriptionTest.java     | 87 ++++++++++++++++++++++
 .../apache/pulsar/client/impl/ConsumerImpl.java    |  4 +
 2 files changed, 91 insertions(+)

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/NonDurableSubscriptionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/NonDurableSubscriptionTest.java
index 6375f79bfbb..b9e3e6dcebe 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/NonDurableSubscriptionTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/NonDurableSubscriptionTest.java
@@ -19,17 +19,29 @@
 package org.apache.pulsar.client.api;
 
 import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNotEquals;
 import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertTrue;
+import java.lang.reflect.Method;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import lombok.Cleanup;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.client.LedgerHandle;
+import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
+import org.apache.pulsar.broker.BrokerTestUtil;
 import org.apache.pulsar.broker.PulsarService;
 import org.apache.pulsar.broker.service.BrokerService;
 import org.apache.pulsar.broker.service.PulsarChannelInitializer;
 import org.apache.pulsar.broker.service.ServerCnx;
+import org.apache.pulsar.broker.service.persistent.PersistentTopic;
 import org.apache.pulsar.client.impl.ConsumerImpl;
+import org.apache.pulsar.client.impl.MessageIdImpl;
 import org.apache.pulsar.common.api.proto.CommandFlow;
+import org.awaitility.Awaitility;
+import org.awaitility.reflect.WhiteboxImpl;
 import org.testng.Assert;
 import org.testng.annotations.AfterMethod;
 import org.testng.annotations.BeforeMethod;
@@ -254,4 +266,79 @@ public class NonDurableSubscriptionTest  extends ProducerConsumerBase {
 
         assertEquals(numFlow.get(), numPartitions);
     }
+
+    private void trimLedgers(final String tpName) {
+        // Wait for topic loading.
+        org.awaitility.Awaitility.await().untilAsserted(() -> {
+            PersistentTopic persistentTopic =
+                    (PersistentTopic) pulsar.getBrokerService().getTopic(tpName, false).join().get();
+            assertNotNull(persistentTopic);
+        });
+        PersistentTopic persistentTopic =
+                (PersistentTopic) pulsar.getBrokerService().getTopic(tpName, false).join().get();
+        ManagedLedgerImpl ml = (ManagedLedgerImpl) persistentTopic.getManagedLedger();
+        CompletableFuture<Void> trimLedgersTask = new CompletableFuture<>();
+        ml.trimConsumedLedgersInBackground(trimLedgersTask);
+        trimLedgersTask.join();
+    }
+
+    private void switchLedgerManually(final String tpName) throws Exception {
+        Method ledgerClosed =
+                ManagedLedgerImpl.class.getDeclaredMethod("ledgerClosed", new Class[]{LedgerHandle.class});
+        Method createLedgerAfterClosed =
+                ManagedLedgerImpl.class.getDeclaredMethod("createLedgerAfterClosed", new Class[0]);
+        ledgerClosed.setAccessible(true);
+        createLedgerAfterClosed.setAccessible(true);
+
+        // Wait for topic create.
+        org.awaitility.Awaitility.await().untilAsserted(() -> {
+            PersistentTopic persistentTopic =
+                    (PersistentTopic) pulsar.getBrokerService().getTopic(tpName, false).join().get();
+            assertNotNull(persistentTopic);
+        });
+
+        // Switch ledger.
+        PersistentTopic persistentTopic =
+                (PersistentTopic) pulsar.getBrokerService().getTopic(tpName, false).join().get();
+        ManagedLedgerImpl ml = (ManagedLedgerImpl) persistentTopic.getManagedLedger();
+        LedgerHandle currentLedger1 = WhiteboxImpl.getInternalState(ml, "currentLedger");
+        ledgerClosed.invoke(ml, new Object[]{currentLedger1});
+        createLedgerAfterClosed.invoke(ml, new Object[0]);
+        Awaitility.await().untilAsserted(() -> {
+            LedgerHandle currentLedger2 = WhiteboxImpl.getInternalState(ml, "currentLedger");
+            assertNotEquals(currentLedger1.getId(), currentLedger2.getId());
+        });
+    }
+
+    @Test
+    public void testTrimLedgerIfNoDurableCursor() throws Exception {
+        final String nonDurableCursor = "non-durable-cursor";
+        final String topicName = BrokerTestUtil.newUniqueName("persistent://public/default/tp");
+        Reader<String> reader = pulsarClient.newReader(Schema.STRING).topic(topicName).receiverQueueSize(1)
+                .subscriptionName(nonDurableCursor).startMessageId(MessageIdImpl.earliest).create();
+        Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(topicName).create();
+        MessageIdImpl msgSent = (MessageIdImpl) producer.send("1");
+
+        // Trigger switch ledger.
+        // Trigger a trim ledgers task, and verify trim ledgers successful.
+        switchLedgerManually(topicName);
+        trimLedgers(topicName);
+
+        // Since there is one message in the incoming queue, so the method "reader.hasMessageAvailable" should return
+        // true.
+        boolean hasMessageAvailable = reader.hasMessageAvailable();
+        Message<String> msgReceived = reader.readNext(2, TimeUnit.SECONDS);
+        if (msgReceived == null) {
+            assertFalse(hasMessageAvailable);
+        } else {
+            log.info("receive msg: {}", msgReceived.getValue());
+            assertTrue(hasMessageAvailable);
+            assertEquals(msgReceived.getValue(), "1");
+        }
+
+        // cleanup.
+        reader.close();
+        producer.close();
+        admin.topics().delete(topicName);
+    }
 }
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
index aa3340c6078..ded6a546c24 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
@@ -2335,6 +2335,10 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
     public CompletableFuture<Boolean> hasMessageAvailableAsync() {
         final CompletableFuture<Boolean> booleanFuture = new CompletableFuture<>();
 
+        if (incomingMessages != null && !incomingMessages.isEmpty()) {
+            return CompletableFuture.completedFuture(true);
+        }
+
         // we haven't read yet. use startMessageId for comparison
         if (lastDequeuedMessageId == MessageId.earliest) {
             // if we are starting from latest, we should seek to the actual last message first.