Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| fe5d474f36 | |||
| d000ac6843 |
@@ -123,7 +123,7 @@ func (s *Subscriber) poll(ctx context.Context) {
|
|||||||
c := s.consumers[tps]
|
c := s.consumers[tps]
|
||||||
s.mu.Unlock()
|
s.mu.Unlock()
|
||||||
if c != nil {
|
if c != nil {
|
||||||
c.recs <- newErrorFetchTopicPartition(kgo.ErrClientClosed, t, p)
|
c.recs <- newErrorFetchTopicPartition(err, t, p)
|
||||||
}
|
}
|
||||||
})
|
})
|
||||||
|
|
||||||
@@ -352,6 +352,8 @@ func (pc *consumer) newErrorMessage(err error, t string, p int32) *kgoMessage {
|
|||||||
pm = &kgoMessage{}
|
pm = &kgoMessage{}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
pm.ack = false
|
||||||
|
pm.body = nil
|
||||||
pm.err = err
|
pm.err = err
|
||||||
pm.topic = t
|
pm.topic = t
|
||||||
pm.hdr = metadata.New(2)
|
pm.hdr = metadata.New(2)
|
||||||
|
|||||||
Reference in New Issue
Block a user