[pulsar] branch branch-2.11 updated: [fix][flaky-test]BatchMessageWithBatchIndexLevelTest.testBatchMessageAck (#17436)

[technoboy]

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

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


The following commit(s) were added to refs/heads/branch-2.11 by this push:
     new db5c1c0036b 
[fix][flaky-test]BatchMessageWithBatchIndexLevelTest.testBatchMessageAck 
(#17436)
db5c1c0036b is described below

commit db5c1c0036bd83e84e41a02103b986bbc31d0028
Author: fengyubiao <yubiao.f...@streamnative.io>
AuthorDate: Tue Sep 6 14:56:01 2022 +0800

    [fix][flaky-test]BatchMessageWithBatchIndexLevelTest.testBatchMessageAck 
(#17436)
---
 .../pulsar/broker/service/BatchMessageWithBatchIndexLevelTest.java    | 4 ++++
 1 file changed, 4 insertions(+)

diff --git 
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BatchMessageWithBatchIndexLevelTest.java
 
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BatchMessageWithBatchIndexLevelTest.java
index b953772ad67..411a53aa821 100644
--- 
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BatchMessageWithBatchIndexLevelTest.java
+++ 
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BatchMessageWithBatchIndexLevelTest.java
@@ -99,11 +99,15 @@ public class BatchMessageWithBatchIndexLevelTest extends 
BatchMessageTest {
         Awaitility.await().untilAsserted(() -> {
             
assertEquals(dispatcher.getConsumers().get(0).getUnackedMessages(), 16);
         });
+        // Block cmd-flow send until verify finish. see: 
https://github.com/apache/pulsar/pull/17436.
+        consumer.pause();
         Message<byte[]> receive5 = consumer.receive();
         consumer.negativeAcknowledge(receive5);
         Awaitility.await().pollInterval(1, 
TimeUnit.MILLISECONDS).untilAsserted(() -> {
             
assertEquals(dispatcher.getConsumers().get(0).getUnackedMessages(), 0);
         });
+        // Unblock cmd-flow.
+        consumer.resume();
         consumer.receive();
         Awaitility.await().untilAsserted(() -> {
             
assertEquals(dispatcher.getConsumers().get(0).getUnackedMessages(), 16);