fix flaky-test: V1_ProducerConsumerTest.testConcurrentConsumerReceiveWhileReconnect
@@ -74,6 +74,7 @@ import org.apache.pulsar.client.impl.TypedMessageBuilderImpl;
import org.apache.pulsar.common.protocol.Commands;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.util.FutureUtil;
+import org.awaitility.Awaitility;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testng.Assert;
@@ -538,11 +539,11 @@ public class V1_ProducerConsumerTest extends V1_ProducerConsumerBase {
String message = "my-message-" + i;
producer.send(message.getBytes());
}
- Thread.sleep(500);
ConsumerImpl<byte[]> consumerImpl = (ConsumerImpl<byte[]>) consumer;
// The available permits should be 10 and num messages in the queue should be 90
- Assert.assertEquals(consumerImpl.getAvailablePermits(), numConsumersThreads);
+ Awaitility.await().untilAsserted(() ->
+ Assert.assertEquals(consumerImpl.getAvailablePermits(), numConsumersThreads));
Assert.assertEquals(consumerImpl.numMessagesInQueue(), recvQueueSize - numConsumersThreads);
barrier.reset();
@@ -557,10 +558,10 @@ public class V1_ProducerConsumerTest extends V1_ProducerConsumerBase {
});
}
barrier.await();
- Thread.sleep(100);
// The available permits should be 20 and num messages in the queue should be 80
- Assert.assertEquals(consumerImpl.getAvailablePermits(), numConsumersThreads * 2);
+ Awaitility.await().untilAsserted(() ->
+ Assert.assertEquals(consumerImpl.getAvailablePermits(), numConsumersThreads * 2));
Assert.assertEquals(consumerImpl.numMessagesInQueue(), recvQueueSize - (numConsumersThreads * 2));
// clear the queue
@@ -591,10 +592,10 @@ public class V1_ProducerConsumerTest extends V1_ProducerConsumerBase {
Thread.sleep(100);
restartBroker();
- Thread.sleep(2000);
// The available permits should be 10 and num messages in the queue should be 90
- Assert.assertEquals(consumerImpl.getAvailablePermits(), numConsumersThreads);
+ Awaitility.await().untilAsserted(() ->
+ Assert.assertEquals(consumerImpl.getAvailablePermits(), numConsumersThreads));
Assert.assertEquals(consumerImpl.numMessagesInQueue(), recvQueueSize - numConsumersThreads);
consumer.close();
