You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/08/04 15:02:28 UTC
(pulsar) 02/08: [fix][test] Await reader reconnect after seek() to fix flaky TopicReaderTest assertions (#26141)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 9430ebe8958881fb0da479661c61882224ef045a
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Fri Jul 3 00:54:49 2026 +0300
[fix][test] Await reader reconnect after seek() to fix flaky TopicReaderTest assertions (#26141)
(cherry picked from commit e0c58b0f9a94f7ba25551255eb61e9938f334edd)
---
.../org/apache/pulsar/client/api/TopicReaderTest.java | 16 ++++++++++------
1 file changed, 10 insertions(+), 6 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TopicReaderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TopicReaderTest.java
index 6e872627fb67..304ddd1fd2f9 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TopicReaderTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TopicReaderTest.java
@@ -55,6 +55,7 @@ import org.apache.pulsar.client.impl.TopicMessageIdImpl;
import org.apache.pulsar.client.impl.TopicMessageImpl;
import org.apache.pulsar.common.policies.data.TopicStats;
import org.apache.pulsar.common.util.RelativeTimeUtil;
+import org.awaitility.Awaitility;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testng.Assert;
@@ -1291,8 +1292,9 @@ public class TopicReaderTest extends SharedPulsarBaseTest {
testMessageOrderAndDuplicates(messageSetB, receivedMessage, expectedMessage);
}
- // Reader should be finished
- assertTrue(reader.isConnected());
+ // Reader should be finished. seek() triggers an asynchronous reconnect of the underlying consumer(s), so
+ // isConnected() can be transiently false right after the post-seek reads; await it to avoid flakiness.
+ Awaitility.await().untilAsserted(() -> assertTrue(reader.isConnected()));
assertFalse(reader.hasMessageAvailable());
assertEquals(((ReaderImpl) reader).getConsumer().numMessagesInQueue(), 0);
@@ -1337,8 +1339,9 @@ public class TopicReaderTest extends SharedPulsarBaseTest {
Assert.assertTrue(messageSetB.add(receivedMessage), "Received duplicate message " + receivedMessage);
}
- // Reader should be finished
- assertTrue(reader.isConnected());
+ // Reader should be finished. seek() triggers an asynchronous reconnect of the underlying consumer(s), so
+ // isConnected() can be transiently false right after the post-seek reads; await it to avoid flakiness.
+ Awaitility.await().untilAsserted(() -> assertTrue(reader.isConnected()));
assertFalse(reader.hasMessageAvailable());
assertEquals(((MultiTopicsReaderImpl) reader).getMultiTopicsConsumer().numMessagesInQueue(), 0);
@@ -1392,8 +1395,9 @@ public class TopicReaderTest extends SharedPulsarBaseTest {
testMessageOrderAndDuplicates(messageSetB, receivedMessage, expectedMessage);
}
- // Reader should be finished
- assertTrue(reader.isConnected());
+ // Reader should be finished. seek() triggers an asynchronous reconnect of the underlying consumer(s), so
+ // isConnected() can be transiently false right after the post-seek reads; await it to avoid flakiness.
+ Awaitility.await().untilAsserted(() -> assertTrue(reader.isConnected()));
assertFalse(reader.hasMessageAvailable());
assertEquals(((ReaderImpl) reader).getConsumer().numMessagesInQueue(), 0);