Fixed flaky test AdminApiTest2.testBacklogNoDelayed
@@ -40,6 +40,7 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.UUID;
import java.util.concurrent.TimeUnit;
import javax.ws.rs.core.Response.Status;
@@ -982,6 +983,7 @@ public class AdminApiTest2 extends MockedPulsarServiceBaseTest {
Consumer<byte[]> consumer = client.newConsumer().topic(topic)
.subscriptionName(subscribeName)
.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
+ .acknowledgmentGroupTime(0, TimeUnit.SECONDS)
.subscribe();
Message<byte[]> message = consumer.receive();
@@ -1077,6 +1079,7 @@ public class AdminApiTest2 extends MockedPulsarServiceBaseTest {
Consumer<byte[]> consumer = client.newConsumer()
.topic(topic)
.subscriptionName(subName)
+ .acknowledgmentGroupTime(0, TimeUnit.SECONDS)
.subscribe();
@Cleanup
@@ -1114,7 +1117,7 @@ public class AdminApiTest2 extends MockedPulsarServiceBaseTest {
@Test(timeOut = 30000)
public void testBacklogNoDelayed() throws PulsarClientException, PulsarAdminException, InterruptedException {
- final String topic = "persistent://prop-xyz/ns1/precise-back-log-no-delayed";
+ final String topic = "persistent://prop-xyz/ns1/precise-back-log-no-delayed-" + UUID.randomUUID().toString();
final String subName = "sub-name";
@Cleanup
@@ -1125,6 +1128,7 @@ public class AdminApiTest2 extends MockedPulsarServiceBaseTest {
.topic(topic)
.subscriptionName(subName)
.subscriptionType(SubscriptionType.Shared)
+ .acknowledgmentGroupTime(0, TimeUnit.SECONDS)
.subscribe();
@Cleanup
@@ -1144,6 +1148,11 @@ public class AdminApiTest2 extends MockedPulsarServiceBaseTest {
}
}
+ // Wait for messages to be tracked for delayed delivery. This happens
+ // on the consumer dispatch side, so when the send() is complete we're
+ // not yet guaranteed to see the stats updated.
+ Thread.sleep(500);
+
TopicStats topicStats = admin.topics().getStats(topic, true);
assertEquals(topicStats.subscriptions.get(subName).msgBacklog, 10);
assertEquals(topicStats.subscriptions.get(subName).msgBacklogNoDelayed, 5);
@@ -1171,6 +1180,7 @@ public class AdminApiTest2 extends MockedPulsarServiceBaseTest {
Consumer<byte[]> consumer = client.newConsumer()
.topic(topic)
.subscriptionName(subName)
+ .acknowledgmentGroupTime(0, TimeUnit.SECONDS)
.subscribe();
@Cleanup
@@ -1212,6 +1222,7 @@ public class AdminApiTest2 extends MockedPulsarServiceBaseTest {
.topic(topic)
.subscriptionName(subName)
.subscriptionType(SubscriptionType.Shared)
+ .acknowledgmentGroupTime(0, TimeUnit.SECONDS)
.subscribe();
@Cleanup
