apache / apache/pulsar-client-go

MetadataError: Consumer not found when instantiating a reader with StartMessageID: pulsar.LatestMessageID()

Open
#452 2 comments 0 reactions 0 assignees View on GitHub
Dominant language
Go
Stars
745
Forks
389
Avg merge
3d 20h
Merged PRs (30d)
3

Description

I am creating a reader to read backwards in time off of a topic, and I have noticed that I receive the above error if I seek by time after creating the reader with `StartMessageID: pulsar.LatestMessageID()` if I use `pulsar.EarliestMessageID()` it seems to behave correctly:

```
../countconsumer localhost:6651 testten testnam testtopic
new client
persistent://testten/testnam/testtopic
new reader
INFO[0000] [Connecting to broker] remote_addr="pulsar://localhost:6651"
INFO[0000] [TCP connection established] local_addr="127.0.0.1:36588" remote_addr="pulsar://localhost:6651"
INFO[0000] [Connection is ready] local_addr="127.0.0.1:36588" remote_addr="pulsar://localhost:6651"
INFO[0000] [Connecting to broker] remote_addr="pulsar://localhost:6651"
INFO[0000] [TCP connection established] local_addr="127.0.0.1:36596" remote_addr="pulsar://localhost:6651"
INFO[0000] [Connection is ready] local_addr="127.0.0.1:36596" remote_addr="pulsar://localhost:6651"
INFO[0000] [Connected consumer] consumerID=1 name= subscription=reader-ruqtj topic="persistent://testten/testnam/testtopic"
INFO[0000] [Created consumer] consumerID=1 name= subscription=reader-ruqtj topic="persistent://testten/testnam/testtopic"
new seek
INFO[0000] Broker notification of Closed consumer: [1] local_addr="127.0.0.1:36596" remote_addr="pulsar://localhost:6651"
INFO[0000] [Reconnecting to broker in 100ms] consumerID=1 name= subscription=reader-ruqtj topic="persistent://testten/testnam/testtopic"
ERRO[0000] [Failed to get last message id] consumerID=1 error="server error: MetadataError: Consumer not found" name= subscription=reader-ruqtj topic="persistent://testten/testnam/testtopic"
ERRO[0000] [Failed to get last message id from broker] error="server error: MetadataError: Consumer not found" topic="persistent://testten/testnam/testtopic"
```

Code attached.

```go
package main

import (
"context"
"fmt"
"log"
"os"
"os/signal"
"strings"
"time"

"github.com/dustin/go-humanize"
"go.uber.org/atomic"

"github.com/apache/pulsar-client-go/pulsar"
)

// TopicName returns a fully-qualified pulsar topic string
func TopicName(tenant, namespace, topic string) string {
return fmt.Sprintf("persistent://%s/%s/%s", tenant, namespace, topic)
}

func main() {
log.SetFlags(0)

log.Println("new client")
client, err := pulsar.NewClient(pulsar.ClientOptions{
URL: "pulsar://" + os.Args[1],
})
if err != nil {
log.Fatal(err)
}
defer client.Close()

topic := TopicName(os.Args[2], os.Args[3], os.Args[4])
log.Println(topic)

log.Println("new reader")
reader, err := client.CreateReader(pulsar.ReaderOptions{
Topic: topic,
StartMessageID: pulsar.LatestMessageID(),
})
if err != nil {
log.Fatal(err)
}
log.Println("new seek")
time.Sleep(100 * time.Millisecond)
reader.SeekByTime(time.Now().Add(-72 * time.Hour))
time.Sleep(100 * time.Millisecond)

ctx, cFunc := context.WithCancel(context.Background())
sig := make(chan os.Signal, 1)
signal.Notify(sig, os.Interrupt, os.Kill)
go func() {
<-sig
cFunc()
}()

var maxPayloadSize int
var count atomic.Uint32
go func() {
t := time.NewTicker(3 * time.Second)
for {
<-t.C
log.Printf("entries: %d\n", count.Load())
}
}()
MainLoop:
for {
select {
case <-ctx.Done():
log.Println("closing gracefully")
reader.Close()
break MainLoop
default:
// get message
}
if reader.HasNext() {
msg, err := reader.Next(ctx)
if err != nil {
log.Println("consumer receive error:", err)
continue
}
count.Add(1)

if len(msg.Payload()) > maxPayloadSize {
maxPayloadSize = len(msg.Payload())
h := humanize.Bytes(uint64(maxPayloadSize))
log.Printf("new max payload size: %d (%s)\n", maxPayloadSize, h)
filename := strings.ReplaceAll(fmt.Sprintf("%s-%s", os.Args[4], h), " ", "")
out, _ := os.Create(filename)
out.Write(msg.Payload())
if err := out.Close(); err != nil {
log.Fatal(err)
}
}

} else {
log.Println("done!")
break MainLoop
}
}
}
```

Contributor guide

Open the contributing guide

Research direction

Start with the CreateReader call using LatestMessageID and the subsequent SeekByTime invocation in the attached reproducer. Trace the reader and consumer lifecycle around the reported Consumer not found error, then verify the behavior against EarliestMessageID. Done means the time seek no longer produces this error and the regression is covered by an appropriate test.

Written by the indexing model from the issue text.

Assessment

Tech stack
go
Domain
distributed-systems
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
35/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.