Fix hasMessageAvailable will return true after seeking to a timestamp newer than the last message (#556)

Author: BewareMyPower

lib/ConsumerImpl.cc

@@ -1642,26 +1642,31 @@ 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, broker will ignore the
+                    // startMessageIdInclusive config, so the compare should still be exclusive
+                    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

@@ -24,8 +24,10 @@
 
 #include <atomic>
 #include <chrono>
+#include <cstdint>
 #include <functional>
 #include <future>
+#include <limits>
 #include <set>
 #include <string>
 #include <thread>
@@ -865,13 +867,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 +979,28 @@ 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()));
+    // Server side (Java) uses signal 64 bits integers to represent the timestamp, so use max int64_t here to
+    // seek to the end of topic.
+    auto now = std::numeric_limits<int64_t>::max();
+    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_.