You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by pe...@apache.org on 2022/07/15 02:36:46 UTC
[pulsar] 15/15: [fix][pulsar-broker] Fix RawReader hasMessageAvailable returns true when no messages (#16443)
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.10
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 0d2a25412a940fe9ea26af484ce7b2331b32be83
Author: Yong Zhang <zh...@gmail.com>
AuthorDate: Thu Jul 14 16:21:28 2022 +0800
[fix][pulsar-broker] Fix RawReader hasMessageAvailable returns true when no messages (#16443)
* [fix][pulsar-broker] Fix RawReader hasMessageAvailable returns true when no messages
---
*Motivation*
The RawReader hasMessageAvailable will return true when all the
messages have been consumed. And that will cause the readNextAsync
blocked and the process never recovered.
In the ConsumerImpl, we update the lastDequeuedMessageId in the
messageProcess. The messageReceived method rewrites by the RawReader,
we should update the lastDequeuedMessageId in the RawReader as well.
* Fix the hasMessageAvaiable in batch messages
(cherry picked from commit 1d174db872914afbafb03607a29d7a2396d87a33)
---
.../apache/pulsar/client/impl/RawReaderImpl.java | 3 ++
.../apache/pulsar/client/impl/RawReaderTest.java | 52 ++++++++++++++++++++++
2 files changed, 55 insertions(+)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawReaderImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawReaderImpl.java
index 92cbbe4a399..331a353b9bd 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawReaderImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawReaderImpl.java
@@ -155,6 +155,9 @@ public class RawReaderImpl implements RawReader {
messageAndCnx.msg.close();
closeAsync();
}
+ MessageIdData messageId = messageAndCnx.msg.getMessageIdData();
+ lastDequeuedMessageId = new BatchMessageIdImpl(messageId.getLedgerId(), messageId.getEntryId(),
+ messageId.getPartition(), numMsg - 1);
ClientCnx currentCnx = cnx();
if (currentCnx == messageAndCnx.cnx) {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RawReaderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RawReaderTest.java
index 5fb3a187f0b..40f36cf4609 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RawReaderTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RawReaderTest.java
@@ -106,6 +106,58 @@ public class RawReaderTest extends MockedPulsarServiceBaseTest {
return msgMetadata.getPartitionKey();
}
+ @Test
+ public void testHasMessageAvailableWithoutBatch() throws Exception {
+ int numKeys = 10;
+ String topic = "persistent://my-property/my-ns/my-raw-topic";
+ Set<String> keys = publishMessages(topic, numKeys);
+ RawReader reader = RawReader.create(pulsarClient, topic, subscription).get();
+ while (true) {
+ boolean hasMsg = reader.hasMessageAvailableAsync().get();
+ if (hasMsg && keys.isEmpty()) {
+ Assert.fail("HasMessageAvailable shows still has message when there is no message");
+ }
+ if (hasMsg) {
+ try (RawMessage m = reader.readNextAsync().get()) {
+ Assert.assertTrue(keys.remove(extractKey(m)));
+ }
+ } else {
+ break;
+ }
+ }
+ Assert.assertTrue(keys.isEmpty());
+ }
+
+ @Test
+ public void testHasMessageAvailableWithBatch() throws Exception {
+ int numKeys = 20;
+ String topic = "persistent://my-property/my-ns/my-raw-topic";
+ Set<String> keys = publishMessages(topic, numKeys, true);
+ RawReader reader = RawReader.create(pulsarClient, topic, subscription).get();
+ int messageCount = 0;
+ while (true) {
+ boolean hasMsg = reader.hasMessageAvailableAsync().get();
+ if (hasMsg && (messageCount == numKeys)) {
+ Assert.fail("HasMessageAvailable shows still has message when there is no message");
+ }
+ if (hasMsg) {
+ try (RawMessage m = reader.readNextAsync().get()) {
+ MessageMetadata meta = Commands.parseMessageMetadata(m.getHeadersAndPayload());
+ messageCount += meta.getNumMessagesInBatch();
+ RawBatchConverter.extractIdsAndKeysAndSize(m).forEach(batchInfo -> {
+ String key = batchInfo.getMiddle();
+ Assert.assertTrue(keys.remove(key));
+ });
+
+ }
+ } else {
+ break;
+ }
+ }
+ Assert.assertEquals(messageCount, numKeys);
+ Assert.assertTrue(keys.isEmpty());
+ }
+
@Test
public void testRawReader() throws Exception {
int numKeys = 10;