[Transaction] Fix flaky txnMessageAckTest.
@@ -100,7 +100,7 @@ public class TransactionEndToEndTest extends TransactionTestBase {
new TenantInfo(Sets.newHashSet("appid1"), Sets.newHashSet(CLUSTER_NAME)));
admin.namespaces().createNamespace(NAMESPACE1);
admin.topics().createPartitionedTopic(TOPIC_OUTPUT, TOPIC_PARTITION);
- admin.topics().createPartitionedTopic(TOPIC_MESSAGE_ACK_TEST, TOPIC_PARTITION);
+ admin.topics().createPartitionedTopic(TOPIC_MESSAGE_ACK_TEST, 1);
admin.tenants().createTenant(NamespaceName.SYSTEM_NAMESPACE.getTenant(),
new TenantInfo(Sets.newHashSet("appid1"), Sets.newHashSet(CLUSTER_NAME)));
@@ -398,7 +398,7 @@ public class TransactionEndToEndTest extends TransactionTestBase {
log.info("produce transaction messages finished");
// Can't receive transaction messages before commit.
- Message<byte[]> message = consumer.receive(5, TimeUnit.SECONDS);
+ Message<byte[]> message = consumer.receive(2, TimeUnit.SECONDS);
Assert.assertNull(message);
log.info("transaction messages can't be received before transaction committed");
@@ -417,10 +417,13 @@ public class TransactionEndToEndTest extends TransactionTestBase {
}
Assert.assertEquals(messageCnt, receiveCnt);
- message = consumer.receive(5, TimeUnit.SECONDS);
+ message = consumer.receive(2, TimeUnit.SECONDS);
Assert.assertNull(message);
- markDeletePositionCheck(topic, subName, false);
+ String checkTopic = TopicName.get(topic).getPartition(0).toString();
+ PersistentTopicInternalStats stats = admin.topics().getInternalStats(checkTopic, false);;
+
+ Assert.assertNotEquals(stats.cursors.get(subName).markDeletePosition, stats.lastConfirmedEntry);
consumer.redeliverUnacknowledgedMessages();
@@ -435,8 +438,8 @@ public class TransactionEndToEndTest extends TransactionTestBase {
message = consumer.receive(2, TimeUnit.SECONDS);
Assert.assertNull(message);
- for (int partition = 0; partition < TOPIC_PARTITION; partition ++) {
- topic = TopicName.get(topic).getPartition(partition).toString();
+
+ topic = TopicName.get(topic).getPartition(0).toString();
boolean exist = false;
for (int i = 0; i < getPulsarServiceList().size(); i++) {
@@ -474,7 +477,6 @@ public class TransactionEndToEndTest extends TransactionTestBase {
}
}
assertTrue(exist);
- }
log.info("receive transaction messages count: {}", receiveCnt);
}
