You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by xy...@apache.org on 2022/04/08 04:42:30 UTC
[pulsar] 01/02: [Client] Add test to ensure the message order in listener callbacks (#15049)
This is an automated email from the ASF dual-hosted git repository.
xyz pushed a commit to branch branch-2.10
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 142dfdb9799c0a6dc79d4351c59559f429d42869
Author: Yunze Xu <xy...@163.com>
AuthorDate: Thu Apr 7 14:55:16 2022 +0800
[Client] Add test to ensure the message order in listener callbacks (#15049)
(cherry picked from commit c2c05c49aff1ebc7b2b7a1d5bd547c33211e4479)
---
.../client/api/SimpleProducerConsumerTest.java | 30 ++++++++++++++++++++++
.../apache/pulsar/client/impl/ConsumerBase.java | 9 +++++--
.../apache/pulsar/client/impl/ConsumerImpl.java | 6 -----
.../client/impl/MultiTopicsConsumerImpl.java | 4 +--
.../pulsar/client/impl/ZeroQueueConsumerImpl.java | 2 +-
5 files changed, 39 insertions(+), 12 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java
index 6eed808f3e4..7506a9bf3f0 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java
@@ -58,6 +58,7 @@ import java.util.Set;
import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutionException;
@@ -4400,4 +4401,33 @@ public class SimpleProducerConsumerTest extends ProducerConsumerBase {
log.info("-- Exiting {} test --", methodName);
}
+
+ @Test(invocationCount = 5)
+ public void testListenerOrdering() throws Exception {
+ final String topic = "persistent://my-property/my-ns/test-listener-ordering-" + System.currentTimeMillis();
+ final int numMessages = 1000;
+ final CountDownLatch latch = new CountDownLatch(numMessages);
+ final List<String> values = new CopyOnWriteArrayList<>();
+ final Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
+ .topic(topic)
+ .subscriptionName("sub")
+ .messageListener((MessageListener<String>) (consumer1, msg) -> {
+ values.add(msg.getValue());
+ latch.countDown();
+ })
+ .subscribe();
+ final Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+ .topic(topic)
+ .create();
+ for (int i = 0; i < numMessages; i++) {
+ producer.send("msg-" + i);
+ }
+ latch.await(3, TimeUnit.SECONDS);
+ producer.close();
+ consumer.close();
+ assertEquals(values.size(), numMessages);
+ for (int i = 0; i < numMessages; i++) {
+ assertEquals(values.get(i), "msg-" + i);
+ }
+ }
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java
index 357da2bdf16..0b2e58988d9 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java
@@ -952,8 +952,13 @@ public abstract class ConsumerBase<T> extends HandlerState implements Consumer<T
}
}
- protected void triggerListener() {
- // Use internalPinnedExecutor to maintain message ordering
+ protected void tryTriggerListener() {
+ if (listener != null) {
+ triggerListener();
+ }
+ }
+
+ private void triggerListener() {
internalPinnedExecutor.execute(() -> {
try {
// Listener should only have one pending/running executable to process a message
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 ff43ab29065..5e319e66755 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
@@ -1344,12 +1344,6 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
}
- private void tryTriggerListener() {
- if (listener != null) {
- triggerListener();
- }
- }
-
private ByteBuf processMessageChunk(ByteBuf compressedPayload, MessageMetadata msgMetadata, MessageIdImpl msgId,
MessageIdData messageId, ClientCnx cnx) {
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
index dc58ec4452d..5898e0fbdf9 100644
--- 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
@@ -307,9 +307,7 @@ public class MultiTopicsConsumerImpl<T> extends ConsumerBase<T> {
notifyPendingBatchReceivedCallBack();
}
- if (listener != null) {
- triggerListener();
- }
+ tryTriggerListener();
}
@Override
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ZeroQueueConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ZeroQueueConsumerImpl.java
index fe6b9294869..8a022f2ab96 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ZeroQueueConsumerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ZeroQueueConsumerImpl.java
@@ -170,7 +170,7 @@ public class ZeroQueueConsumerImpl<T> extends ConsumerImpl<T> {
}
@Override
- protected void triggerListener() {
+ protected void tryTriggerListener() {
// Ignore since it was already triggered in the triggerZeroQueueSizeListener() call
}