Fix Reader HasMessageAvailable behavior after seeking by timestamp (#344)

* Fix Reader HasMessageAvailable behavior after seeking by timestamp

* Improve logic

Author: RobertIndie

src/Pulsar.Client/Internal/ConsumerImpl.fs

@@ -1097,6 +1097,7 @@ type internal ConsumerImpl<'T> (consumerConfig: ConsumerConfiguration<'T>, clien
                             hasSoughtByTimestamp <- true
                             Commands.newSeekByTimestamp consumerId requestId timestamp, MessageId.Earliest
                         | SeekType.MessageId messageId ->
+                            hasSoughtByTimestamp <- false
                             match messageId.ChunkMessageIds with
                             | Some chunkMessageIds when chunkMessageIds.Length >0 ->
                                     Commands.newSeekByMsgId consumerId requestId chunkMessageIds[0], chunkMessageIds[0]
@@ -1145,8 +1146,9 @@ type internal ConsumerImpl<'T> (consumerConfig: ConsumerConfiguration<'T>, clien
                                 let! lastMessageIdResult = getLastMessageIdAsync()
                                 let lastMessageId = lastMessageIdResult.LastMessageId
                                 // if the consumer is configured to read inclusive then we need to seek to the last message
-                                do! postAndAsyncReply this.Mb (fun channel ->
-                                                SeekAsync (SeekType.MessageId lastMessageId, channel))
+                                if consumerConfig.ResetIncludeHead && not hasSoughtByTimestamp then
+                                    do! postAndAsyncReply this.Mb (fun channel ->
+                                                    SeekAsync (SeekType.MessageId lastMessageId, channel))
                                 match lastMessageIdResult.MarkDeletePosition with
                                 | Some markDeletePosition ->
                                     if lastMessageId.EntryId < %0L then

tests/IntegrationTests/Reader.fs

@@ -350,6 +350,53 @@ let tests =
 
             Log.Debug("Finished HasMessageAvailableAfterSeekTimestamp initializeLastMessageIdInBroker: {0}", initializeLastMessageIdInBroker)
         }
+
+    let checkReaderLoopWhileHasMessageAvailableAfterSeekTimestamp () =
+        task {
+            Log.Debug("Started Reader loop while HasMessageAvailable after SeekTimestamp")
+            let client = getClient()
+            let topicName = "public/default/test-reader-loop-has-message-available-after-seek-timestamp-" + Guid.NewGuid().ToString("N")
+
+            let! (producer : IProducer<string>) =
+                client.NewProducer(Schema.STRING())
+                    .Topic(topicName)
+                    .EnableBatching(false)
+                    .CreateAsync()
+
+            let timestampBeforeSend = %(DateTimeOffset.UtcNow.ToUnixTimeMilliseconds())
+            let! _ = producer.SendAsync("m1")
+            let! _ = producer.SendAsync("m2")
+            let! _ = producer.SendAsync("m3")
+            do! producer.DisposeAsync()
+
+            let! (reader : IReader<string>) =
+                client.NewReader(Schema.STRING())
+                    .Topic(topicName)
+                    .ReceiverQueueSize(1)
+                    .StartMessageId(MessageId.Latest)
+                    .CreateAsync()
+
+            do! Task.Delay(1000)
+            do! reader.SeekAsync(timestampBeforeSend)
+            do! Task.Delay(1000)
+
+            use cts = new System.Threading.CancellationTokenSource(TimeSpan.FromSeconds(10.0))
+            let received = ResizeArray<string>()
+
+            let mutable continueLooping = true
+            while continueLooping do
+                let! hasMessage = reader.HasMessageAvailableAsync()
+                if hasMessage then
+                    let! msg = reader.ReadNextAsync(cts.Token)
+                    received.Add(msg.GetValue())
+                else
+                    continueLooping <- false
+
+            Expect.sequenceEqual "" [ "m1"; "m2"; "m3" ] received
+            do! reader.DisposeAsync()
+
+            Log.Debug("Finished Reader loop while HasMessageAvailable after SeekTimestamp")
+        }
 
     testList "Reader" [
 
@@ -408,4 +455,8 @@ let tests =
         testTask "HasMessageAvailable after SeekTimestamp with initializeLastMessageIdInBroker" {
             do! checkHasMessageAvailableAfterSeekTimestamp true 
         }
+
+        testTask "Check HasMessageAvailable works after SeekTimestamp" {
+            do! checkReaderLoopWhileHasMessageAvailableAfterSeekTimestamp()
+        }
     ]