[ISSUE-328] test if there is no message

Signed-off-by: Paulo Pereira <[email protected]>

pulsar/consumer_partition.go

@@ -100,6 +100,10 @@ const (
 	nonDurable
 )
 
+const (
+	noMessageEntry = -1
+)
+
 type partitionConsumerOpts struct {
 	topic                      string
 	consumerName               string
@@ -198,11 +202,13 @@ func newPartitionConsumer(parent Consumer, client *client, options *partitionCon
 		if err != nil {
 			return nil, err
 		}
-		pc.startMessageID = msgID
+		if msgID.entryID != noMessageEntry {
+			pc.startMessageID = msgID
 
-		err = pc.requestSeek(msgID)
-		if err != nil {
-			return nil, err
+			err = pc.requestSeek(msgID)
+			if err != nil {
+				return nil, err
+			}
 		}
 	}
 

pulsar/reader_test.go

@@ -458,6 +458,18 @@ func TestReaderLatestInclusiveHasNext(t *testing.T) {
 	topic := newTopicName()
 	ctx := context.Background()
 
+	// create reader on the last message (inclusive)
+	reader0, err := client.CreateReader(ReaderOptions{
+		Topic:                   topic,
+		StartMessageID:          LatestMessageID(),
+		StartMessageIDInclusive: true,
+	})
+
+	assert.Nil(t, err)
+	defer reader0.Close()
+
+	assert.False(t, reader0.HasNext())
+
 	// create producer
 	producer, err := client.CreateProducer(ProducerOptions{
 		Topic:           topic,