fixup revoked error

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>

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)
 				}
 			})
 
@@ -199,7 +199,15 @@ func (s *Subscriber) revoked(ctx context.Context, c *kgo.Client, revoked map[str
 	}
 	s.killConsumers(ctx, revoked)
 	if err := c.CommitMarkedOffsets(ctx); err != nil {
-		s.kopts.Logger.Error(ctx, "[kgo] revoked CommitMarkedOffsets error", err)
+		s.mu.Lock()
+		tpc := make(map[tp]*consumer, len(s.consumers))
+		maps.Copy(tpc, s.consumers)
+		s.mu.Unlock()
+		for tp, c := range tpc {
+			if c != nil {
+				c.recs <- newErrorFetchTopicPartition(err, tp.t, tp.p)
+			}
+		}
 	}
 }
 
@@ -352,6 +360,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)