apache / apache/pulsar-client-go
MetadataError: Consumer not found when instantiating a reader with StartMessageID: pulsar.LatestMessageID()
- 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
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