Fix : ConsumerStatsTest.testAsyncCallOnPartitionedTopic is flaky (#5066)

fix #4894

### Motivation

Due to the concurrent execution of the program, when checking whether the ‘BrokerConsumerStats’'s cache is in effect, there is no guarantee that the cache time will not be exceeded.

### Modifications

- Check the cache in an inexact way.
- Increase waiting time

pulsar-client-cpp/tests/ConsumerStatsTest.cc

@@ -40,12 +40,16 @@ static std::string lookupUrl = "pulsar://localhost:6650";
 static std::string adminUrl = "http://localhost:8080/";
 
 void partitionedCallbackFunction(Result result, BrokerConsumerStats brokerConsumerStats, long expectedBacklog,
-                                 Latch& latch, int index) {
+                                 Latch& latch, int index, bool accurate) {
     ASSERT_EQ(result, ResultOk);
     PartitionedBrokerConsumerStatsImpl* statsPtr =
         (PartitionedBrokerConsumerStatsImpl*)(brokerConsumerStats.getImpl().get());
     LOG_DEBUG(statsPtr);
-    ASSERT_EQ(expectedBacklog, statsPtr->getBrokerConsumerStats(index).getMsgBacklog());
+    if (accurate) {
+        ASSERT_EQ(expectedBacklog, statsPtr->getBrokerConsumerStats(index).getMsgBacklog());
+    } else {
+        ASSERT_LE(expectedBacklog, statsPtr->getBrokerConsumerStats(index).getMsgBacklog());
+    }
     latch.countdown();
 }
 
@@ -283,8 +287,8 @@ TEST(ConsumerStatsTest, testAsyncCallOnPartitionedTopic) {
 
     // Expecting return from 4 callbacks
     Latch latch(4);
-    consumer.getBrokerConsumerStatsAsync(
-        std::bind(partitionedCallbackFunction, std::placeholders::_1, std::placeholders::_2, 5, latch, 0));
+    consumer.getBrokerConsumerStatsAsync(std::bind(partitionedCallbackFunction, std::placeholders::_1,
+                                                   std::placeholders::_2, 5, latch, 0, true));
 
     // Now we have 10 messages per partition
     for (int i = numOfMessages; i < (numOfMessages * 2); i++) {
@@ -294,23 +298,25 @@ TEST(ConsumerStatsTest, testAsyncCallOnPartitionedTopic) {
     }
 
     // Expecting cached result
-    consumer.getBrokerConsumerStatsAsync(
-        std::bind(partitionedCallbackFunction, std::placeholders::_1, std::placeholders::_2, 5, latch, 0));
+    // Inaccurate judgment is used because it cannot guarantee that the above operations are completed within
+    // cache time.
+    consumer.getBrokerConsumerStatsAsync(std::bind(partitionedCallbackFunction, std::placeholders::_1,
+                                                   std::placeholders::_2, 5, latch, 0, false));
 
     std::this_thread::sleep_for(std::chrono::milliseconds(4500));
     // Expecting fresh results
-    consumer.getBrokerConsumerStatsAsync(
-        std::bind(partitionedCallbackFunction, std::placeholders::_1, std::placeholders::_2, 10, latch, 2));
+    consumer.getBrokerConsumerStatsAsync(std::bind(partitionedCallbackFunction, std::placeholders::_1,
+                                                   std::placeholders::_2, 10, latch, 2, true));
 
     Message msg;
     while (consumer.receive(msg)) {
         // Do nothing
     }
 
     // Expecting the backlog to be the same since we didn't acknowledge the messages
-    consumer.getBrokerConsumerStatsAsync(
-        std::bind(partitionedCallbackFunction, std::placeholders::_1, std::placeholders::_2, 10, latch, 3));
+    consumer.getBrokerConsumerStatsAsync(std::bind(partitionedCallbackFunction, std::placeholders::_1,
+                                                   std::placeholders::_2, 10, latch, 3, true));
 
     // Wait for ten seconds only
-    ASSERT_TRUE(latch.wait(std::chrono::seconds(10)));
+    ASSERT_TRUE(latch.wait(std::chrono::seconds(30)));
 }