From 5ceb05663eb552e92c9ff6ea388f342ac23ffacd Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Thu, 28 Sep 2023 23:45:39 +0800 Subject: [PATCH] [fix] [client] fix reader.hasMessageAvailable return false when incoming queue is not empty (#21259) 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`. Make `reader.hasMessageAvailable` return `true` when `incoming queue` is not empty. (cherry picked from commit 6d82b09128f46fdcb27021560d773fac15d66a48) (cherry picked from commit 38c3f0c019fdf3cb8f1dcf0c739226ebea777bc7) --- .../api/NonDurableSubscriptionTest.java | 89 ++++++++++++++++++- .../pulsar/client/impl/ConsumerImpl.java | 4 + 2 files changed, 92 insertions(+), 1 deletion(-) 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 6375f79bfbb6e..0f19da1c0ec93 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.AssertJUnit.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; @@ -38,7 +50,7 @@ @Test(groups = "broker-api") @Slf4j -public class NonDurableSubscriptionTest extends ProducerConsumerBase { +public class NonDurableSubscriptionTest extends ProducerConsumerBase { private final AtomicInteger numFlow = new AtomicInteger(0); @@ -254,4 +266,79 @@ public void testFlowCountForMultiTopics() throws Exception { 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 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 reader = pulsarClient.newReader(Schema.STRING).topic(topicName).receiverQueueSize(1) + .subscriptionName(nonDurableCursor).startMessageId(MessageIdImpl.earliest).create(); + Producer 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 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 fa5a8bf3c5aa4..d71b410d7afda 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 @@ -2339,6 +2339,10 @@ public boolean hasMessageAvailable() throws PulsarClientException { public CompletableFuture hasMessageAvailableAsync() { final CompletableFuture 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.