apache / apache/pulsar-client-go

consume stuck after seekByTiime

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

Description

#### Expected behavior
consume success after seekByTime

#### Actual behavior

After running the for loop a few times, step msg := <-consumer.Chan() will be blocked.

#### Steps to reproduce

```go
func TestPulsarChannel2(t *testing.T) {
Params.Init()
c, err := pulsar.NewClient(pulsar.ClientOptions{URL: Params.PulsarAddress})
assert.Nil(t, err)
for i := 0; i < 100; i++ {
channel := strconv.Itoa(rand.Int())

p, err := c.CreateProducer(pulsar.ProducerOptions{Topic: channel})
assert.Nil(t, err)
ppm := &pulsar.ProducerMessage{Payload: []byte{1}, Properties: map[string]string{}}
_, err = p.Send(context.TODO(), ppm)
assert.Nil(t, err)

p.Close()

consumer, err := c.Subscribe(pulsar.ConsumerOptions{
Topic: channel,
SubscriptionName: "test",
Type: pulsar.KeyShared,
SubscriptionInitialPosition: pulsar.SubscriptionPositionEarliest,
})
assert.Nil(t, err)
consumer.SeekByTime(time.Unix(0, 0))
msg := <-consumer.Chan()
assert.EqualValues(t, []byte{1}, msg.Payload)

consumer.Close()
}
}
```

#### System configuration
**Pulsar version**: x.y

Contributor guide

Open the contributing guide

Research direction

Start by running the reported TestPulsarChannel2 reproduction and trace the interaction between consumer.SeekByTime(time.Unix(0, 0)) and consumer.Chan(). Done means the loop completes without blocking and the received message payload remains []byte{1}.

Written by the indexing model from the issue text.

Assessment

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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.