apache / apache/pulsar-client-go
consume stuck after seekByTiime
- 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
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