[pulsar] branch branch-2.9 updated: [fix][flaky-test]ManagedCursorMetricsTest.testCursorReadWriteMetrics (#17045)

[bogong]

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

bogong pushed a commit to branch branch-2.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git


The following commit(s) were added to refs/heads/branch-2.9 by this push:
     new 73c6755754d 
[fix][flaky-test]ManagedCursorMetricsTest.testCursorReadWriteMetrics (#17045)
73c6755754d is described below

commit 73c6755754d4bbc9eb8572f650fbbe069fb31eb0
Author: fengyubiao <yubiao.f...@streamnative.io>
AuthorDate: Fri Aug 12 20:25:38 2022 +0800

    [fix][flaky-test]ManagedCursorMetricsTest.testCursorReadWriteMetrics 
(#17045)
    
    (cherry picked from commit e0ff3d7e0083d4ecd9b4af55e11a6bf9e461dfc9)
---
 .../broker/stats/ManagedCursorMetricsTest.java     | 24 +++++++++++++++++++---
 1 file changed, 21 insertions(+), 3 deletions(-)

diff --git 
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/ManagedCursorMetricsTest.java
 
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/ManagedCursorMetricsTest.java
index 812a6b2e047..6b5f002ab41 100644
--- 
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/ManagedCursorMetricsTest.java
+++ 
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/ManagedCursorMetricsTest.java
@@ -21,6 +21,7 @@ package org.apache.pulsar.broker.stats;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.UUID;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import lombok.Cleanup;
 import org.apache.bookkeeper.client.LedgerHandle;
@@ -200,9 +201,19 @@ public class ManagedCursorMetricsTest extends 
MockedPulsarServiceBaseTest {
         admin.topics().delete(topicName, true);
     }
 
+    private ManagedCursorMXBean getManagedCursorMXBean(String topicName, 
String subscriptionName)
+            throws ExecutionException, InterruptedException {
+        final PersistentSubscription persistentSubscription =
+                (PersistentSubscription) pulsar.getBrokerService()
+                        .getTopic(topicName, 
false).get().get().getSubscription(subscriptionName);
+        final ManagedCursorImpl managedCursor = (ManagedCursorImpl) 
persistentSubscription.getCursor();
+        return managedCursor.getStats();
+    }
+
     @Test
     public void testCursorReadWriteMetrics() throws Exception {
-        final String subName = "read-write";
+        final String subName1 = "read-write-sub-1";
+        final String subName2 = "read-write-sub-2";
         final String topicName = 
"persistent://my-namespace/use/my-ns/read-write";
         final int messageSize = 10;
 
@@ -219,7 +230,7 @@ public class ManagedCursorMetricsTest extends 
MockedPulsarServiceBaseTest {
                 .topic(topicName)
                 .subscriptionType(SubscriptionType.Shared)
                 .ackTimeout(1, TimeUnit.SECONDS)
-                .subscriptionName(subName)
+                .subscriptionName(subName1)
                 .subscribe();
 
         @Cleanup
@@ -227,7 +238,7 @@ public class ManagedCursorMetricsTest extends 
MockedPulsarServiceBaseTest {
                 .topic(topicName)
                 .subscriptionType(SubscriptionType.Shared)
                 .ackTimeout(1, TimeUnit.SECONDS)
-                .subscriptionName(subName + "-2")
+                .subscriptionName(subName2)
                 .subscribe();
 
         @Cleanup
@@ -244,6 +255,13 @@ public class ManagedCursorMetricsTest extends 
MockedPulsarServiceBaseTest {
                 consumer2.acknowledge(consumer.receive().getMessageId());
             }
         }
+
+        // Wait for persistent cursor meta.
+        ManagedCursorMXBean cursorMXBean1 = getManagedCursorMXBean(topicName, 
subName1);
+        ManagedCursorMXBean cursorMXBean2 = getManagedCursorMXBean(topicName, 
subName2);
+        Awaitility.await().until(() -> 
cursorMXBean1.getWriteCursorLedgerLogicalSize() > 0);
+        Awaitility.await().until(() -> 
cursorMXBean2.getWriteCursorLedgerLogicalSize() > 0);
+
         metricsList = metrics.generate();
         Assert.assertEquals(metricsList.size(), 2);
         
Assert.assertEquals(metricsList.get(0).getMetrics().get("brk_ml_cursor_writeLedgerSize"),
 26L);