pass errors from broker to subscribers

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2025-12-02 22:08:09 +03:00
parent bac736deef
commit c8eeb34efe
5 changed files with 195 additions and 39 deletions
--- a/kgo.go
+++ b/kgo.go
@@ -489,6 +489,7 @@ func (k *Broker) Subscribe(ctx context.Context, topic string, handler broker.Han
 		kgo.OnPartitionsLost(sub.lost),
 		kgo.AutoCommitCallback(sub.autocommit),
 		kgo.AutoCommitMarks(),
+		kgo.WithHooks(sub),
 	)

 	if options.Context != nil {