cornelk commented on issue #328:
URL: 
https://github.com/apache/pulsar-client-go/issues/328#issuecomment-674397141


   @quintans I had similar problems with this functionality in the past, I 
recreated this test in my alternative Pulsar-Go Client ( 
https://github.com/cornelk/pulsar-go ) to analyze the problem:
   
   ``` go
   package main
   
   import (
        "context"
        "fmt"
        "log"
        "os"
        "testing"
   
        "github.com/cornelk/pulsar-go"
        "github.com/stretchr/testify/assert"
        "github.com/stretchr/testify/require"
   )
   
   const (
        lookupURL = "pulsar://localhost:6650"
        topic     = "persistent://public/default/test"
   )
   
   func TestReaderLatestInclusiveHasNext(t *testing.T) {
        l := log.New(os.Stderr, "", log.LstdFlags)
        client, err := pulsar.NewClient(lookupURL, pulsar.WithLogger(l))
        require.Nil(t, err)
        defer client.Close()
   
        ctx := context.Background()
        err = client.Dial(ctx)
        require.Nil(t, err)
   
        // create producer
        producer, err := client.NewProducer(ctx, pulsar.ProducerConfig{
                Topic:     topic,
                BatchSize: 1,
        })
        assert.Nil(t, err)
        defer producer.Close()
   
        // send 10 messages
        var lastMsgID *pulsar.MessageID
        for i := 0; i < 10; i++ {
                lastMsgID, err = producer.WriteMessage(ctx, 
[]byte(fmt.Sprintf("hello-%d", i)))
                assert.NoError(t, err)
                assert.NotNil(t, lastMsgID)
        }
   
        // create reader on the last message (inclusive)
        reader, err := client.NewConsumer(ctx, pulsar.ConsumerConfig{
                Topic:                   topic,
                InitialPosition:         pulsar.LatestPosition,
                StartMessageIDInclusive: true,
        })
   
        require.Nil(t, err)
        defer reader.Close()
   
        msg, err := reader.ReadMessage(context.Background())
        require.NoError(t, err)
   
        assert.Equal(t, []byte("hello-9"), msg.Body) // updated after original 
post
        msgID := msg.ID
   
        require.NotNil(t, msgID)
        id1, err := lastMsgID.Marshal()
        id2, err := msgID.Marshal()
        assert.Equal(t, id1, id2) // updated after original post
   }
   ```
   
   The problem with your test code and my client was that after 
starting/subscribing the consumer, Pulsar was not done sending back the last 
message yet, which makes the `HasNext()` call fail. The solution for my client 
was to directly call `ReadMessage()`, which blocks until Pulsar sends the 
message back.


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
[email protected]


Reply via email to