[Fix][Flaky-test] Fix testConsumeTxnMessage
* [Fix][Flaky-test] Fix testConsumeTxnMessage
Master
## Motivation
The transaction commit is async, so the consumer can still receive message when the consumer rebuilds.
## Modification
Add  Awaitility.await() for check-ing whether the ongoingTxns = 0.
@@ -37,6 +37,7 @@ import org.apache.pulsar.common.naming.SystemTopicNames;
import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
import org.apache.pulsar.common.policies.data.ClusterData;
import org.apache.pulsar.common.policies.data.TenantInfoImpl;
+import org.awaitility.Awaitility;
import org.testng.Assert;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
@@ -196,7 +197,7 @@ public class PerformanceTransactionTest extends MockedPulsarServiceBaseTest {
}
@Test
- public void testConsumeTxnMessage() throws InterruptedException, PulsarClientException {
+ public void testConsumeTxnMessage() throws Exception {
String argString = "%s -r 10 -u %s -txn -ss %s -st %s -sp %s -ntxn %d";
String subName = "sub";
String topic = testTopic + UUID.randomUUID();
@@ -224,6 +225,13 @@ public class PerformanceTransactionTest extends MockedPulsarServiceBaseTest {
});
thread.start();
thread.join();
+
+ Awaitility.await().untilAsserted(() -> {
+ admin.transactions().getCoordinatorStats().forEach((integer, transactionCoordinatorStats) -> {
+ Assert.assertEquals(transactionCoordinatorStats.ongoingTxnSize, 0);
+ });
+ });
+
Consumer<byte[]> consumer = pulsarClient.newConsumer().subscriptionName(subName).topic(topic)
.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
.subscriptionType(SubscriptionType.Exclusive)
