From 7007d2bce0a2eb7b260675a9e7e36b4593bef9b4 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Nicol=C3=B2=20Boschi?= <boschi1997@gmail.com>
Date: Wed, 17 Aug 2022 16:40:17 +0200
Subject: [PATCH 1/2] [fix][test] Fix flaky test
 TopicReaderTest.testMultiReaderIsAbleToSeekWithTimeOnMiddleOfTopic

---
 .../apache/pulsar/client/api/TopicReaderTest.java  | 14 +++++++++-----
 1 file changed, 9 insertions(+), 5 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 8b533b5b450c5..09c1fc0338b51 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
@@ -1383,26 +1383,30 @@ public void testReaderIsAbleToSeekWithTimeOnMiddleOfTopic() throws Exception {
     public void testMultiReaderIsAbleToSeekWithTimeOnMiddleOfTopic() throws Exception {
         final String topicName = "persistent://my-property/my-ns/testMultiReaderIsAbleToSeekWithTimeOnMiddleOfTopic" + System.currentTimeMillis();
         final int numOfMessage = 10;
-        final int halfMessages = numOfMessage / 2;
         admin.topics().createPartitionedTopic(topicName, 3);
         Producer<byte[]> producer = pulsarClient.newProducer().topic(topicName).create();
         long halfTime = 0;
         for (int i = 0; i < numOfMessage; i++) {
-            if (i == numOfMessage / 2) {
+            if (i == 6) {
                 halfTime = System.currentTimeMillis();
             }
             producer.send(String.format("msg num %d", i).getBytes());
         }
         Assert.assertTrue(halfTime != 0);
-        Reader<byte[]> reader = pulsarClient.newReader().topic(topicName).startMessageId(MessageId.earliest).create();
+        Reader<byte[]> reader = pulsarClient.newReader().topic(topicName).startMessageId(MessageId.latest).create();
 
         reader.seek(halfTime);
         Set<String> messageSet = Sets.newHashSet();
-        for (int i = halfMessages + 1; i < numOfMessage; i++) {
-            Message<byte[]> message = reader.readNext(10, TimeUnit.SECONDS);
+        while (true) {
+            Message<byte[]> message = reader.readNext(1, TimeUnit.SECONDS);
+            // make sure we are headed to the end of the topic
+            if (message == null) {
+                break;
+            }
             String receivedMessage = new String(message.getData());
             Assert.assertTrue(messageSet.add(receivedMessage), "Received duplicate message " + receivedMessage);
         }
+        assertEquals(messageSet.size(), 4);
         reader.close();
         producer.close();
     }

From 6b5ef389fbd597f3301da5e1add112d82757bef6 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Nicol=C3=B2=20Boschi?= <boschi1997@gmail.com>
Date: Thu, 18 Aug 2022 08:41:26 +0200
Subject: [PATCH 2/2] also fix testReaderIsAbleToSeekWithTimeOnMiddleOfTopic

---
 .../pulsar/client/api/TopicReaderTest.java    | 24 ++++++++++++-------
 1 file changed, 15 insertions(+), 9 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 09c1fc0338b51..e02f1dd0e4619 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
@@ -1350,30 +1350,36 @@ public void testReaderIsAbleToSeekWithMessageIdOnMiddleOfTopic() throws Exceptio
     public void testReaderIsAbleToSeekWithTimeOnMiddleOfTopic() throws Exception {
         final String topicName = "persistent://my-property/my-ns/ReaderIsAbleToSeekWithTimeOnMiddleOfTopic";
         final int numOfMessage = 10;
-        final int halfMessages = numOfMessage / 2;
 
         Producer<byte[]> producer = pulsarClient.newProducer()
                 .topic(topicName).create();
 
-        long l = System.currentTimeMillis();
+        long halfTime = 0;
         for (int i = 0; i < numOfMessage; i++) {
+            if (i == 6) {
+                halfTime = System.currentTimeMillis();
+            }
             producer.send(String.format("msg num %d", i).getBytes());
-            Thread.sleep(100);
         }
 
         Reader<byte[]> reader = pulsarClient.newReader().topic(topicName)
-                .startMessageId(MessageId.earliest).create();
+                .startMessageId(MessageId.latest).create();
 
-        int plusTime = (halfMessages + 1) * 100;
-        reader.seek(l + plusTime);
+        reader.seek(halfTime);
 
         Set<String> messageSet = Sets.newHashSet();
-        for (int i = halfMessages + 1; i < numOfMessage; i++) {
-            Message<byte[]> message = reader.readNext();
+        int i = 6;
+        while (true) {
+            Message<byte[]> message = reader.readNext(1, TimeUnit.SECONDS);
+            // make sure we are headed to the end of the topic
+            if (message == null) {
+                break;
+            }
             String receivedMessage = new String(message.getData());
-            String expectedMessage = String.format("msg num %d", i);
+            String expectedMessage = String.format("msg num %d", i++);
             testMessageOrderAndDuplicates(messageSet, receivedMessage, expectedMessage);
         }
+        assertEquals(messageSet.size(), 4);
 
         reader.close();
         producer.close();