fixup error

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
pass errors from broker to subscribers
2025-12-02 21:46:03 +03:00 · 2025-12-02 21:35:35 +03:00
1 changed files with 3 additions and 1 deletions
--- a/subscriber.go
+++ b/subscriber.go
@@ -123,7 +123,7 @@ func (s *Subscriber) poll(ctx context.Context) {
 				c := s.consumers[tps]
 				s.mu.Unlock()
 				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.ack = false
+	pm.body = nil
 	pm.err = err
 	pm.topic = t
 	pm.hdr = metadata.New(2)
Author	SHA1	Message	Date
Vasiliy Tolstov	fe5d474f36	fixup error Some checks failed coverage / build (push) Has been cancelled Details test / test (push) Has been cancelled Details sync / sync (push) Successful in 11s Details Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>	2025-12-02 21:46:03 +03:00
Vasiliy Tolstov	d000ac6843	pass errors from broker to subscribers Some checks are pending coverage / build (push) Waiting to run Details test / test (push) Waiting to run Details sync / sync (push) Has started running Details Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>	2025-12-02 21:35:35 +03:00