Compare commits

..

1 Commits

Author SHA1 Message Date
5080a52834 fixup revoked error
Some checks failed
test / test (push) Failing after 12m26s
coverage / build (push) Failing after 18m17s
sync / sync (push) Successful in 7s
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2025-12-02 21:50:53 +03:00

View File

@@ -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)
}
}
}
}