You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by zh...@apache.org on 2023/02/27 06:26:38 UTC
[pulsar] 01/04: [fix][client] Fix reader listener can't auto ack with pooled message. (#19354)
This is an automated email from the ASF dual-hosted git repository.
zhaocong pushed a commit to branch branch-2.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 42261c67a6bc94fefe6750e667535f79812d0581
Author: Qiang Zhao <ma...@apache.org>
AuthorDate: Mon Jan 30 13:42:23 2023 +0800
[fix][client] Fix reader listener can't auto ack with pooled message. (#19354)
(cherry picked from commit 1b2fa1f2dafc39b5f2bd7e17cc6130066dbc8aad)
---
.../org/apache/pulsar/client/impl/ReaderTest.java | 113 +++++++++++++++++++++
.../pulsar/client/impl/MultiTopicsReaderImpl.java | 7 +-
.../org/apache/pulsar/client/impl/ReaderImpl.java | 7 +-
3 files changed, 125 insertions(+), 2 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ReaderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ReaderTest.java
index 53ae8f97008..6289f5ed784 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ReaderTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ReaderTest.java
@@ -26,6 +26,7 @@ import static org.testng.Assert.fail;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import java.io.IOException;
+import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashSet;
@@ -55,6 +56,7 @@ import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.ClusterData;
import org.apache.pulsar.common.policies.data.ManagedLedgerInternalStats;
+import org.apache.pulsar.common.policies.data.PartitionedTopicInternalStats;
import org.apache.pulsar.common.policies.data.PersistentTopicInternalStats;
import org.apache.pulsar.common.policies.data.RetentionPolicies;
import org.apache.pulsar.common.policies.data.TenantInfoImpl;
@@ -621,4 +623,115 @@ public class ReaderTest extends MockedPulsarServiceBaseTest {
Assert.assertEquals(internalStats.cursors.size(), 0);
}
+ @Test
+ public void testReaderListenerAcknowledgement()
+ throws IOException, InterruptedException, PulsarAdminException {
+ // non-partitioned topic
+ final String topic = "persistent://my-property/my-ns/" + UUID.randomUUID();
+ admin.topics().createNonPartitionedTopic(topic);
+ final Producer<byte[]> producer = pulsarClient.newProducer()
+ .topic(topic)
+ .create();
+ producer.send("0".getBytes(StandardCharsets.UTF_8));
+ // non-pool
+ final CountDownLatch readerNonPoolLatch = new CountDownLatch(1);
+ final Reader<byte[]> readerNonPool = pulsarClient.newReader()
+ .topic(topic)
+ .subscriptionName("reader-non-pool")
+ .startMessageId(MessageId.earliest)
+ .readerListener((innerReader, message) -> {
+ // no operation
+ readerNonPoolLatch.countDown();
+ }).create();
+ readerNonPoolLatch.await();
+ Awaitility.await().untilAsserted(() -> {
+ final PersistentTopicInternalStats internal = admin.topics().getInternalStats(topic);
+ final String lastConfirmedEntry = internal.lastConfirmedEntry;
+ Assert.assertTrue(internal.cursors.containsKey("reader-non-pool"));
+ Assert.assertEquals(internal.cursors.get("reader-non-pool").markDeletePosition, lastConfirmedEntry);
+ });
+ // pooled
+ final CountDownLatch readerPooledLatch = new CountDownLatch(1);
+ final Reader<byte[]> readerPooled = pulsarClient.newReader()
+ .topic(topic)
+ .subscriptionName("reader-pooled")
+ .startMessageId(MessageId.earliest)
+ .poolMessages(true)
+ .readerListener((innerReader, message) -> {
+ try {
+ // no operation
+ readerPooledLatch.countDown();
+ } finally {
+ message.release();
+ }
+ }).create();
+ readerPooledLatch.await();
+ Awaitility.await().untilAsserted(() -> {
+ final PersistentTopicInternalStats internal = admin.topics().getInternalStats(topic);
+ final String lastConfirmedEntry = internal.lastConfirmedEntry;
+ Assert.assertTrue(internal.cursors.containsKey("reader-pooled"));
+ Assert.assertEquals(internal.cursors.get("reader-pooled").markDeletePosition, lastConfirmedEntry);
+ });
+ producer.close();
+ readerNonPool.close();
+ readerPooled.close();
+ admin.topics().delete(topic);
+ // ---- partitioned topic
+ final String partitionedTopic = "persistent://my-property/my-ns/" + UUID.randomUUID();
+ admin.topics().createPartitionedTopic(partitionedTopic, 2);
+ final Producer<byte[]> producer2 = pulsarClient.newProducer()
+ .topic(partitionedTopic)
+ .create();
+ producer2.send("0".getBytes(StandardCharsets.UTF_8));
+ // non-pool
+ final CountDownLatch readerNonPoolLatch2 = new CountDownLatch(1);
+ final Reader<byte[]> readerNonPool2 = pulsarClient.newReader()
+ .topic(partitionedTopic)
+ .subscriptionName("reader-non-pool")
+ .startMessageId(MessageId.earliest)
+ .readerListener((innerReader, message) -> {
+ // no operation
+ readerNonPoolLatch2.countDown();
+ }).create();
+ readerNonPoolLatch2.await();
+ Awaitility.await().untilAsserted(() -> {
+ PartitionedTopicInternalStats partitionedInternal =
+ admin.topics().getPartitionedInternalStats(partitionedTopic);
+ for (PersistentTopicInternalStats internal : partitionedInternal.partitions.values()) {
+ final String lastConfirmedEntry = internal.lastConfirmedEntry;
+ Assert.assertTrue(internal.cursors.containsKey("reader-non-pool"));
+ Assert.assertEquals(internal.cursors.get("reader-non-pool").markDeletePosition, lastConfirmedEntry);
+ }
+ });
+ // pooled
+ final CountDownLatch readerPooledLatch2 = new CountDownLatch(1);
+ final Reader<byte[]> readerPooled2 = pulsarClient.newReader()
+ .topic(partitionedTopic)
+ .subscriptionName("reader-pooled")
+ .startMessageId(MessageId.earliest)
+ .poolMessages(true)
+ .readerListener((innerReader, message) -> {
+ try {
+ // no operation
+ readerPooledLatch2.countDown();
+ } finally {
+ message.release();
+ }
+ }).create();
+ readerPooledLatch2.await();
+ Awaitility.await().untilAsserted(() -> {
+ PartitionedTopicInternalStats partitionedInternal =
+ admin.topics().getPartitionedInternalStats(partitionedTopic);
+ for (PersistentTopicInternalStats internal : partitionedInternal.partitions.values()) {
+ final String lastConfirmedEntry = internal.lastConfirmedEntry;
+ Assert.assertTrue(internal.cursors.containsKey("reader-pooled"));
+ Assert.assertEquals(internal.cursors.get("reader-pooled").markDeletePosition, lastConfirmedEntry);
+ }
+ });
+ producer2.close();
+ readerNonPool2.close();
+ readerPooled2.close();
+ admin.topics().deletePartitionedTopic(partitionedTopic);
+ }
+
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsReaderImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsReaderImpl.java
index b656c005db9..0d4dd87d2c2 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsReaderImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsReaderImpl.java
@@ -73,8 +73,13 @@ public class MultiTopicsReaderImpl<T> implements Reader<T> {
@Override
public void received(Consumer<T> consumer, Message<T> msg) {
+ final MessageId messageId = msg.getMessageId();
readerListener.received(MultiTopicsReaderImpl.this, msg);
- consumer.acknowledgeCumulativeAsync(msg);
+ consumer.acknowledgeCumulativeAsync(messageId).exceptionally(ex -> {
+ log.error("[{}][{}] auto acknowledge message {} cumulative fail.", getTopic(),
+ getMultiTopicsConsumer().getSubscription(), messageId, ex);
+ return null;
+ });
}
@Override
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ReaderImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ReaderImpl.java
index 72c74e8875d..64ee3c9c8b5 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ReaderImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ReaderImpl.java
@@ -90,8 +90,13 @@ public class ReaderImpl<T> implements Reader<T> {
@Override
public void received(Consumer<T> consumer, Message<T> msg) {
+ final MessageId messageId = msg.getMessageId();
readerListener.received(ReaderImpl.this, msg);
- consumer.acknowledgeCumulativeAsync(msg);
+ consumer.acknowledgeCumulativeAsync(messageId).exceptionally(ex -> {
+ log.error("[{}][{}] auto acknowledge message {} cumulative fail.", getTopic(),
+ getConsumer().getSubscription(), messageId, ex);
+ return null;
+ });
}
@Override