Fix flaky test in org.apache.pulsar.client.api.KeySharedSubscriptionTest#testRemoveFirstConsumer
@@ -23,9 +23,7 @@ import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
-
import com.google.common.collect.Sets;
-
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Paths;
@@ -43,15 +41,14 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
-
import lombok.Cleanup;
-
import org.apache.bookkeeper.mledger.impl.PositionImpl;
import org.apache.curator.shaded.com.google.common.collect.Lists;
import org.apache.pulsar.broker.service.Topic;
import org.apache.pulsar.broker.service.nonpersistent.NonPersistentStickyKeyDispatcherMultipleConsumers;
import org.apache.pulsar.broker.service.persistent.PersistentStickyKeyDispatcherMultipleConsumers;
import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
+import org.apache.pulsar.client.impl.ConsumerImpl;
import org.apache.pulsar.common.api.proto.KeySharedMode;
import org.apache.pulsar.common.naming.TopicDomain;
import org.apache.pulsar.common.naming.TopicName;
@@ -670,6 +667,8 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase {
}
// All the already published messages will be pre-fetched by C1.
+ Awaitility.await().untilAsserted(() ->
+ assertEquals(((ConsumerImpl<Integer>) c1).getTotalIncomingMessages(), 10));
// Adding a new consumer.
@Cleanup
