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

[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 458ff57f235 
[fix][flaky-test]BatchMessageWithBatchIndexLevelTest.testBatchMessageAck 
(#17436)
458ff57f235 is described below

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

    [fix][flaky-test]BatchMessageWithBatchIndexLevelTest.testBatchMessageAck 
(#17436)
    
    (cherry picked from commit 427e129a4d72d846a0757d5e5e7ad40afa19e0a5)
---
 .../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 d5c4e1eb064..731c65dd33d 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
@@ -101,11 +101,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);