Fix hasMessageAvailable will return true after seeking to a timestamp newer than the last message

lib/ConsumerImpl.cc

@@ -1642,26 +1642,33 @@ void ConsumerImpl::hasMessageAvailableAsync(const HasMessageAvailableCallback& c
                 callback(result, {});
                 return;
             }
-            auto handleResponse = [self, response, callback] {
+            bool lastSeekIsByTimestamp = false;
+            {
+                LockGuard lock{self->mutex_};
+                if (self->lastSeekArg_.has_value() &&
+                    std::holds_alternative<SeekTimestampType>(self->lastSeekArg_.value())) {
+                    lastSeekIsByTimestamp = true;
+                }
+            }
+            auto handleResponse = [self, lastSeekIsByTimestamp, response, callback] {
                 if (response.hasMarkDeletePosition() && response.getLastMessageId().entryId() >= 0) {
                     // We only care about comparing ledger ids and entry ids as mark delete position
                     // doesn't have other ids such as batch index
                     auto compareResult = compareLedgerAndEntryId(response.getMarkDeletePosition(),
                                                                  response.getLastMessageId());
-                    callback(ResultOk, self->config_.isStartMessageIdInclusive() ? compareResult <= 0
-                                                                                 : compareResult < 0);
+                    // When the consumer has sought by timestamp that is later than the last message, the
+                    // mark-delete position will still be the same with the last message id's position. But
+                    // broker won't dispatch messages even if startMessageId is inclusive, so we should return
+                    // false in this case.
+                    if (lastSeekIsByTimestamp || !self->config_.isStartMessageIdInclusive()) {
+                        callback(ResultOk, compareResult < 0);
+                    } else {
+                        callback(ResultOk, compareResult <= 0);
+                    }
                 } else {
                     callback(ResultOk, false);
                 }
             };
-            bool lastSeekIsByTimestamp = false;
-            {
-                LockGuard lock{self->mutex_};
-                if (self->lastSeekArg_.has_value() &&
-                    std::holds_alternative<SeekTimestampType>(self->lastSeekArg_.value())) {
-                    lastSeekIsByTimestamp = true;
-                }
-            }
             if (self->config_.isStartMessageIdInclusive() && !lastSeekIsByTimestamp) {
                 self->seekAsync(response.getLastMessageId(), [callback, handleResponse](Result result) {
                     if (result != ResultOk) {

tests/ReaderTest.cc

@@ -38,6 +38,7 @@
 #include "lib/Latch.h"
 #include "lib/LogUtils.h"
 #include "lib/ReaderImpl.h"
+#include "lib/TimeUtils.h"
 DECLARE_LOG_OBJECT()
 
 using namespace pulsar;
@@ -865,13 +866,7 @@ TEST_P(ReaderSeekTest, testHasMessageAvailableAfterSeekToEnd) {
     }
 
     ASSERT_EQ(ResultOk, reader.seek(MessageId::latest()));
-    // After seek-to-end the broker may close the consumer and trigger reconnect; allow a short
-    // delay for hasMessageAvailable to become false (avoids flakiness when reconnect completes).
-    for (int i = 0; i < 50; i++) {
-        ASSERT_EQ(ResultOk, reader.hasMessageAvailable(hasMessageAvailable));
-        if (!hasMessageAvailable) break;
-        std::this_thread::sleep_for(std::chrono::milliseconds(100));
-    }
+    ASSERT_EQ(ResultOk, reader.hasMessageAvailable(hasMessageAvailable));
     ASSERT_FALSE(hasMessageAvailable);
 
     producer.send(MessageBuilder().setContent("msg-2").build());
@@ -983,6 +978,26 @@ TEST_F(ReaderSeekTest, testSeekInclusiveChunkMessage) {
     assertStartMessageId(false, secondMsgId);
 }
 
+TEST_P(ReaderSeekTest, testSeekToEndByTimestamp) {
+    auto topic = "test-seek-to-end-by-timestamp-" + std::to_string(time(nullptr));
+    Producer producer;
+    ASSERT_EQ(ResultOk, client.createProducer(topic, producer));
+
+    ReaderConfiguration readerConf;
+    readerConf.setStartMessageIdInclusive(GetParam());
+
+    Reader reader;
+    ASSERT_EQ(ResultOk, client.createReader(topic, MessageId::earliest(), readerConf, reader));
+
+    ASSERT_EQ(ResultOk, producer.send(MessageBuilder().setContent("msg").build()));
+    auto now = TimeUtils::currentTimeMillis() + 1000;
+    ASSERT_EQ(ResultOk, reader.seek(now));
+
+    bool hasMessageAvailable;
+    ASSERT_EQ(ResultOk, reader.hasMessageAvailable(hasMessageAvailable));
+    ASSERT_FALSE(hasMessageAvailable);
+}
+
 // Regression test for segfault when Reader is used with messageListenerThreads=0.
 // Verifies ExecutorServiceProvider(0) does not cause undefined behavior and
 // ConsumerImpl::messageReceived does not dereference null listenerExecutor_.