[pulsar] branch master updated: [fix][test] Fix flaky test NonPersistentTopicTest.testMsgDropStat, follow up (#20401)

[mm...@apache.org]

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

mmarshall 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 25c4b7cee40 [fix][test] Fix flaky test NonPersistentTopicTest.testMsgDropStat, follow up (#20401)
25c4b7cee40 is described below

commit 25c4b7cee402a1d486d720a71dc2e06aa6d9af64
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Thu May 25 19:28:18 2023 +0300

    [fix][test] Fix flaky test NonPersistentTopicTest.testMsgDropStat, follow up (#20401)
    
    Fixes #20386
    
    ### Motivation
    
    - the previous attempt #20387 to fix the flakiness wasn't effective
    
    ### Modifications
    
    - improve the fix and rely on the fact that entryId is -1 when the message is dropped in the broker
      code: https://github.com/apache/pulsar/blob/091ee2504ffbe6ec98e354b76e7f4c045e1914aa/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java#L1699-L1711
    
    ### Documentation
    
    <!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
    
    - [ ] `doc` <!-- Your PR contains doc changes. -->
    - [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
    - [x] `doc-not-needed` <!-- Your PR changes do not impact docs -->
    - [ ] `doc-complete` <!-- Docs have been already added -->
---
 .../pulsar/client/api/NonPersistentTopicTest.java  | 30 +++++++++++++++-------
 1 file changed, 21 insertions(+), 9 deletions(-)

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/NonPersistentTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/NonPersistentTopicTest.java
index c41ab3e8ccc..63ce0f00dff 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/NonPersistentTopicTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/NonPersistentTopicTest.java
@@ -38,6 +38,7 @@ import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
 import lombok.Cleanup;
 import org.apache.pulsar.broker.PulsarService;
 import org.apache.pulsar.broker.ServiceConfiguration;
@@ -50,6 +51,7 @@ import org.apache.pulsar.broker.service.nonpersistent.NonPersistentReplicator;
 import org.apache.pulsar.broker.service.nonpersistent.NonPersistentTopic;
 import org.apache.pulsar.client.admin.PulsarAdmin;
 import org.apache.pulsar.client.impl.ConsumerImpl;
+import org.apache.pulsar.client.impl.MessageIdImpl;
 import org.apache.pulsar.client.impl.MultiTopicsConsumerImpl;
 import org.apache.pulsar.client.impl.PartitionedProducerImpl;
 import org.apache.pulsar.client.impl.ProducerImpl;
@@ -824,13 +826,14 @@ public class NonPersistentTopicTest extends ProducerConsumerBase {
             conf.setMaxConcurrentNonPersistentMessagePerConnection(1);
             stopBroker();
             startBroker();
+
+            pulsar.getBrokerService().updateRates();
+
             Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topicName).subscriptionName("subscriber-1")
-                    .receiverQueueSize(1)
-                    .messageListener((c, msg) -> {}).subscribe();
+                    .receiverQueueSize(1).subscribe();
 
             Consumer<byte[]> consumer2 = pulsarClient.newConsumer().topic(topicName).subscriptionName("subscriber-2")
-                    .receiverQueueSize(1).subscriptionType(SubscriptionType.Shared)
-                    .messageListener((c, msg) -> {}).subscribe();
+                    .receiverQueueSize(1).subscriptionType(SubscriptionType.Shared).subscribe();
 
             ProducerImpl<byte[]> producer = (ProducerImpl<byte[]>) pulsarClient.newProducer().topic(topicName)
                 .enableBatching(false)
@@ -839,17 +842,26 @@ public class NonPersistentTopicTest extends ProducerConsumerBase {
             @Cleanup("shutdownNow")
             ExecutorService executor = Executors.newFixedThreadPool(5);
             byte[] msgData = "testData".getBytes();
-            final int totalProduceMessages = 200;
-            CountDownLatch latch = new CountDownLatch(totalProduceMessages);
+            final int totalProduceMessages = 1000;
+            CountDownLatch latch = new CountDownLatch(1);
+            AtomicInteger messagesSent = new AtomicInteger(0);
             for (int i = 0; i < totalProduceMessages; i++) {
                 executor.submit(() -> {
-                    producer.sendAsync(msgData).handle((msg, e) -> {
-                        latch.countDown();
+                    producer.sendAsync(msgData).handle((msgId, e) -> {
+                        int count = messagesSent.incrementAndGet();
+                        // process at least 20% of messages before signalling the latch
+                        // a non-persistent message will return entryId as -1 when it has been dropped
+                        // due to setMaxConcurrentNonPersistentMessagePerConnection limit
+                        // also ensure that it has happened before the latch is signalled
+                        if (count > totalProduceMessages * 0.2 && msgId != null
+                                && ((MessageIdImpl) msgId).getEntryId() == -1) {
+                            latch.countDown();
+                        }
                         return null;
                     });
                 });
             }
-            latch.await();
+            latch.await(5, TimeUnit.SECONDS);
 
             NonPersistentTopic topic =
                     (NonPersistentTopic) pulsar.getBrokerService().getOrCreateTopic(topicName).get();