pass errors from broker to subscribers

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

go.mod

@@ -7,17 +7,17 @@ toolchain go1.24.3
 require (
 	github.com/google/uuid v1.6.0
 	github.com/stretchr/testify v1.11.1
-	github.com/twmb/franz-go v1.19.5
-	github.com/twmb/franz-go/pkg/kadm v1.16.1
-	github.com/twmb/franz-go/pkg/kmsg v1.11.2
+	github.com/twmb/franz-go v1.20.2
+	github.com/twmb/franz-go/pkg/kadm v1.17.1
+	github.com/twmb/franz-go/pkg/kmsg v1.12.0
 	go.opentelemetry.io/otel v1.38.0
-	go.unistack.org/micro/v3 v3.11.45
+	go.unistack.org/micro/v3 v3.11.48
 )
 
 require (
 	github.com/ash3in/uuidv8 v1.2.0 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
-	github.com/klauspost/compress v1.18.0 // indirect
+	github.com/klauspost/compress v1.18.1 // indirect
 	github.com/kr/pretty v0.3.1 // indirect
 	github.com/matoous/go-nanoid v1.5.1 // indirect
 	github.com/pierrec/lz4/v4 v4.1.22 // indirect
@@ -25,7 +25,7 @@ require (
 	go.unistack.org/micro-proto/v3 v3.4.1 // indirect
 	golang.org/x/crypto v0.43.0 // indirect
 	golang.org/x/sys v0.37.0 // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20251007200510-49b9836ed3ff // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20251022142026-3a174f9686a8 // indirect
 	google.golang.org/grpc v1.76.0 // indirect
 	google.golang.org/protobuf v1.36.10 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect

go.sum

@@ -13,6 +13,8 @@ github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
 github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo=
 github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ=
+github.com/klauspost/compress v1.18.1 h1:bcSGx7UbpBqMChDtsF28Lw6v/G94LPrrbMbdC3JH2co=
+github.com/klauspost/compress v1.18.1/go.mod h1:ZQFFVG+MdnR0P+l6wpXgIL4NTtwiKIdBnrBd8Nrxr+0=
 github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
 github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
@@ -31,16 +33,24 @@ github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu
 github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
 github.com/twmb/franz-go v1.19.5 h1:W7+o8D0RsQsedqib71OVlLeZ0zI6CbFra7yTYhZTs5Y=
 github.com/twmb/franz-go v1.19.5/go.mod h1:4kFJ5tmbbl7asgwAGVuyG1ZMx0NNpYk7EqflvWfPCpM=
+github.com/twmb/franz-go v1.20.2 h1:CiwhyKZHW6vqSHJkh+RTxFAJkio0jBjM/JQhx/HZ72A=
+github.com/twmb/franz-go v1.20.2/go.mod h1:YCnepDd4gl6vdzG03I5Wa57RnCTIC6DVEyMpDX/J8UA=
 github.com/twmb/franz-go/pkg/kadm v1.16.1 h1:IEkrhTljgLHJ0/hT/InhXGjPdmWfFvxp7o/MR7vJ8cw=
 github.com/twmb/franz-go/pkg/kadm v1.16.1/go.mod h1:Ue/ye1cc9ipsQFg7udFbbGiFNzQMqiH73fGC2y0rwyc=
+github.com/twmb/franz-go/pkg/kadm v1.17.1 h1:Bt02Y/RLgnFO2NP2HVP1kd2TFtGRiJZx+fSArjZDtpw=
+github.com/twmb/franz-go/pkg/kadm v1.17.1/go.mod h1:s4duQmrDbloVW9QTMXhs6mViTepze7JLG43xwPcAeTg=
 github.com/twmb/franz-go/pkg/kmsg v1.11.2 h1:hIw75FpwcAjgeyfIGFqivAvwC5uNIOWRGvQgZhH4mhg=
 github.com/twmb/franz-go/pkg/kmsg v1.11.2/go.mod h1:CFfkkLysDNmukPYhGzuUcDtf46gQSqCZHMW1T4Z+wDE=
+github.com/twmb/franz-go/pkg/kmsg v1.12.0 h1:CbatD7ers1KzDNgJqPbKOq0Bz/WLBdsTH75wgzeVaPc=
+github.com/twmb/franz-go/pkg/kmsg v1.12.0/go.mod h1:+DPt4NC8RmI6hqb8G09+3giKObE6uD2Eya6CfqBpeJY=
 go.opentelemetry.io/otel v1.38.0 h1:RkfdswUDRimDg0m2Az18RKOsnI8UDzppJAtj01/Ymk8=
 go.opentelemetry.io/otel v1.38.0/go.mod h1:zcmtmQ1+YmQM9wrNsTGV/q/uyusom3P8RxwExxkZhjM=
 go.unistack.org/micro-proto/v3 v3.4.1 h1:UTjLSRz2YZuaHk9iSlVqqsA50JQNAEK2ZFboGqtEa9Q=
 go.unistack.org/micro-proto/v3 v3.4.1/go.mod h1:okx/cnOhzuCX0ggl/vToatbCupi0O44diiiLLsZ93Zo=
 go.unistack.org/micro/v3 v3.11.45 h1:fjTLZYWgsVf9FIMZBxOg8ios2/tmyimnjZrsrxEUeXU=
 go.unistack.org/micro/v3 v3.11.45/go.mod h1:fDQ8Mu9wubaFP0L8hNQlpzHiEnWN0wbOlawN9HYo0N4=
+go.unistack.org/micro/v3 v3.11.48 h1:lHJYSHU2z1TTcuswItGwG7cZXN6n04EFqY7lk/0gA7w=
+go.unistack.org/micro/v3 v3.11.48/go.mod h1:fDQ8Mu9wubaFP0L8hNQlpzHiEnWN0wbOlawN9HYo0N4=
 golang.org/x/crypto v0.43.0 h1:dduJYIi3A3KOfdGOHX8AVZ/jGiyPa3IbBozJ5kNuE04=
 golang.org/x/crypto v0.43.0/go.mod h1:BFbav4mRNlXJL4wNeejLpWxB7wMbc79PdRGhWKncxR0=
 golang.org/x/net v0.45.0 h1:RLBg5JKixCy82FtLJpeNlVM0nrSqpCRYzVU1n8kj0tM=
@@ -51,6 +61,8 @@ golang.org/x/text v0.30.0 h1:yznKA/E9zq54KzlzBEAWn1NXSQ8DIp/NYMy88xJjl4k=
 golang.org/x/text v0.30.0/go.mod h1:yDdHFIX9t+tORqspjENWgzaCVXgk0yYnYuSZ8UzzBVM=
 google.golang.org/genproto/googleapis/rpc v0.0.0-20251007200510-49b9836ed3ff h1:A90eA31Wq6HOMIQlLfzFwzqGKBTuaVztYu/g8sn+8Zc=
 google.golang.org/genproto/googleapis/rpc v0.0.0-20251007200510-49b9836ed3ff/go.mod h1:7i2o+ce6H/6BluujYR+kqX3GKH+dChPTQU19wjRPiGk=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20251022142026-3a174f9686a8 h1:M1rk8KBnUsBDg1oPGHNCxG4vc1f49epmTO7xscSajMk=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20251022142026-3a174f9686a8/go.mod h1:7i2o+ce6H/6BluujYR+kqX3GKH+dChPTQU19wjRPiGk=
 google.golang.org/grpc v1.76.0 h1:UnVkv1+uMLYXoIz6o7chp59WfQUYA2ex/BXQ9rHZu7A=
 google.golang.org/grpc v1.76.0/go.mod h1:Ju12QI8M6iQJtbcsV+awF5a4hfJMLi4X0JLo94ULZ6c=
 google.golang.org/protobuf v1.36.10 h1:AYd7cD/uASjIL6Q9LiTjz8JLcrh/88q5UObnmY3aOOE=

hook_event.go

@@ -17,12 +17,12 @@ type hookEvent struct {
 }
 
 var (
-	_ kgo.HookBrokerConnect           = &hookEvent{}
-	_ kgo.HookBrokerDisconnect        = &hookEvent{}
-	_ kgo.HookBrokerRead              = &hookEvent{}
-	_ kgo.HookBrokerWrite             = &hookEvent{}
-	_ kgo.HookGroupManageError        = &hookEvent{}
-	_ kgo.HookProduceRecordUnbuffered = &hookEvent{}
+	_ kgo.HookBrokerConnect           = (*hookEvent)(nil)
+	_ kgo.HookBrokerDisconnect        = (*hookEvent)(nil)
+	_ kgo.HookBrokerRead              = (*hookEvent)(nil)
+	_ kgo.HookBrokerWrite             = (*hookEvent)(nil)
+	_ kgo.HookGroupManageError        = (*hookEvent)(nil)
+	_ kgo.HookProduceRecordUnbuffered = (*hookEvent)(nil)
 )
 
 func (m *hookEvent) OnGroupManageError(err error) {

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 {

kgo_test.go

@@ -17,7 +17,7 @@ import (
 )
 
 var (
-	msgcnt   = int64(1200)
+	msgcnt   = int64(10)
 	group    = "38"
 	prefill  = true
 	loglevel = logger.DebugLevel
@@ -80,7 +80,7 @@ func TestFail(t *testing.T) {
 	for _, msg := range msgs {
 		t.Logf("broker publish")
 		if err := b.Publish(ctx, "test", msg); err != nil {
-			t.Fatal(err)
+			break
 		}
 	}
 	//	t.Skip()
@@ -108,7 +108,7 @@ func TestFail(t *testing.T) {
 
 	for {
 		t.Logf("health check")
-		if !b.Health() {
+		if b.Health() {
 			t.Logf("health works")
 			break
 		}

meter.go

@@ -14,45 +14,45 @@ type hookMeter struct {
 }
 
 var (
-	_ kgo.HookBrokerConnect    = &hookMeter{}
-	_ kgo.HookBrokerDisconnect = &hookMeter{}
+	_ kgo.HookBrokerConnect    = (*hookMeter)(nil)
+	_ kgo.HookBrokerDisconnect = (*hookMeter)(nil)
 	// HookBrokerE2E
-	_ kgo.HookBrokerRead     = &hookMeter{}
-	_ kgo.HookBrokerThrottle = &hookMeter{}
-	_ kgo.HookBrokerWrite    = &hookMeter{}
-	_ kgo.HookFetchBatchRead = &hookMeter{}
+	_ kgo.HookBrokerRead     = (*hookMeter)(nil)
+	_ kgo.HookBrokerThrottle = (*hookMeter)(nil)
+	_ kgo.HookBrokerWrite    = (*hookMeter)(nil)
+	_ kgo.HookFetchBatchRead = (*hookMeter)(nil)
 	// HookFetchRecordBuffered
 	// HookFetchRecordUnbuffered
-	_ kgo.HookGroupManageError = &hookMeter{}
+	_ kgo.HookGroupManageError = (*hookMeter)(nil)
 	// HookNewClient
-	_ kgo.HookProduceBatchWritten = &hookMeter{}
+	_ kgo.HookProduceBatchWritten = (*hookMeter)(nil)
 	// HookProduceRecordBuffered
 	// HookProduceRecordPartitioned
 	// HookProduceRecordUnbuffered
 )
 
 const (
-	metricBrokerConnects    = "broker_connects_total"
-	metricBrokerDisconnects = "broker_disconnects_total"
+	metricBrokerConnects    = "micro_broker_connects_total"
+	metricBrokerDisconnects = "micro_broker_disconnects_total"
 
-	metricBrokerWriteErrors        = "broker_write_errors_total"
-	metricBrokerWriteBytes         = "broker_write_bytes_total"
-	metricBrokerWriteWaitLatencies = "broker_write_wait_latencies"
-	metricBrokerWriteLatencies     = "broker_write_latencies"
+	metricBrokerWriteErrors        = "micro_broker_write_errors_total"
+	metricBrokerWriteBytes         = "micro_broker_write_bytes_total"
+	metricBrokerWriteWaitLatencies = "micro_broker_write_wait_latencies"
+	metricBrokerWriteLatencies     = "micro_broker_write_latencies"
 
-	metricBrokerReadErrors        = "broker_read_errors_total"
-	metricBrokerReadBytes         = "broker_read_bytes_total"
-	metricBrokerReadWaitLatencies = "broker_read_wait_latencies"
-	metricBrokerReadLatencies     = "broker_read_latencies"
+	metricBrokerReadErrors        = "micro_broker_read_errors_total"
+	metricBrokerReadBytes         = "micro_broker_read_bytes_total"
+	metricBrokerReadWaitLatencies = "micro_broker_read_wait_latencies"
+	metricBrokerReadLatencies     = "micro_broker_read_latencies"
 
-	metricBrokerThrottleLatencies = "broker_throttle_latencies"
+	metricBrokerThrottleLatencies = "micro_broker_throttle_latencies"
 
-	metricBrokerProduceBytesCompressed   = "broker_produce_bytes_compressed_total"
-	metricBrokerProduceBytesUncompressed = "broker_produce_bytes_uncompressed_total"
-	metricBrokerFetchBytesCompressed     = "broker_consume_bytes_compressed_total"
-	metricBrokerFetchBytesUncompressed   = "broker_consume_bytes_uncompressed_total"
+	metricBrokerProduceBytesCompressed   = "micro_broker_produce_bytes_compressed_total"
+	metricBrokerProduceBytesUncompressed = "micro_broker_produce_bytes_uncompressed_total"
+	metricBrokerFetchBytesCompressed     = "micro_broker_consume_bytes_compressed_total"
+	metricBrokerFetchBytesUncompressed   = "micro_broker_consume_bytes_uncompressed_total"
 
-	metricBrokerGroupErrors = "broker_group_errors_total"
+	metricBrokerGroupErrors = "micro_broker_group_errors_total"
 
 	labelNode    = "node_id"
 	labelSuccess = "success"

subscriber.go

@@ -3,6 +3,9 @@ package kgo
 import (
 	"context"
 	"fmt"
+	"maps"
+	"net"
+	"strconv"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -59,7 +62,7 @@ type Subscriber struct {
 	closed       bool
 	fatalOnError bool
 
-	sync.RWMutex
+	mu sync.RWMutex
 }
 
 func (s *Subscriber) Client() *kgo.Client {
@@ -113,13 +116,24 @@ func (s *Subscriber) poll(ctx context.Context) {
 			fetches := s.c.PollRecords(ctx, maxInflight)
 			if !s.closed && fetches.IsClientClosed() {
 				s.closed = true
+				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(kgo.ErrClientClosed, tp.t, tp.p)
+					}
+				}
 				return
 			}
 			fetches.EachError(func(t string, p int32, err error) {
-				if kgo.IsRetryableBrokerErr(err) {
-					s.kopts.Logger.Error(ctx, fmt.Sprintf("[kgo] fetch topic %s partition %d error", t, p), err)
-				} else {
-					s.kopts.Logger.Fatal(ctx, fmt.Sprintf("[kgo] fetch topic %s partition %d error", t, p), err)
+				tps := tp{t, p}
+				s.mu.Lock()
+				c := s.consumers[tps]
+				s.mu.Unlock()
+				if c != nil {
+					c.recs <- newErrorFetchTopicPartition(err, t, p)
 				}
 			})
 
@@ -140,7 +154,7 @@ func (s *Subscriber) killConsumers(ctx context.Context, lost map[string][]int32)
 		for _, partition := range partitions {
 			tps := tp{topic, partition}
 			pc, ok := s.consumers[tps]
-			if !ok {
+			if !ok || pc == nil {
 				continue
 			}
 			delete(s.consumers, tps)
@@ -154,11 +168,20 @@ func (s *Subscriber) killConsumers(ctx context.Context, lost map[string][]int32)
 	}
 }
 
-func (s *Subscriber) autocommit(_ *kgo.Client, _ *kmsg.OffsetCommitRequest, _ *kmsg.OffsetCommitResponse, err error) {
+func (s *Subscriber) autocommit(_ *kgo.Client, r *kmsg.OffsetCommitRequest, _ *kmsg.OffsetCommitResponse, err error) {
 	if err != nil {
-		//		s.connected.Store(0)
-		if s.fatalOnError {
-			s.kopts.Logger.Fatal(context.TODO(), "kgo.AutoCommitCallback error", err)
+		s.mu.Lock()
+		tpc := make(map[tp]*consumer, len(s.consumers))
+		maps.Copy(tpc, s.consumers)
+		s.mu.Unlock()
+		for _, tc := range r.Topics {
+			for _, c := range s.consumers {
+				if c != nil {
+					for _, p := range tc.Partitions {
+						c.recs <- newErrorFetchTopicPartition(err, tc.Topic, p.Partition)
+					}
+				}
+			}
 		}
 	}
 }
@@ -168,7 +191,6 @@ func (s *Subscriber) lost(ctx context.Context, _ *kgo.Client, lost map[string][]
 		s.kopts.Logger.Error(ctx, fmt.Sprintf("[kgo] lost %#+v", lost))
 	}
 	s.killConsumers(ctx, lost)
-	// s.connected.Store(0)
 }
 
 func (s *Subscriber) revoked(ctx context.Context, c *kgo.Client, revoked map[string][]int32) {
@@ -177,8 +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.connected.Store(0)
+		s.mu.Lock()
+		tpc := make(map[tp]*consumer, len(s.consumers))
+		maps.Copy(tpc, s.consumers)
+		s.mu.Unlock()
+		for tp, c := range s.consumers {
+			if c != nil {
+				c.recs <- newErrorFetchTopicPartition(err, tp.t, tp.p)
+			}
+		}
 	}
 }
 
@@ -198,9 +227,9 @@ func (s *Subscriber) assigned(_ context.Context, c *kgo.Client, assigned map[str
 				opts:      s.opts,
 				connected: s.connected,
 			}
-			s.Lock()
+			s.mu.Lock()
 			s.consumers[tp{topic, partition}] = pc
-			s.Unlock()
+			s.mu.Unlock()
 			go pc.consume()
 		}
 	}
@@ -218,11 +247,24 @@ func (pc *consumer) consume() {
 		eh = pc.opts.ErrorHandler
 	}
 
+	var pm *event
+
 	for {
 		select {
 		case <-pc.quit:
 			return
 		case p := <-pc.recs:
+			if p.Err != nil || p.FetchPartition.Err != nil {
+				if p.Err != nil {
+					pm = pc.newErrorMessage(p.Err, p.Topic, p.Partition)
+				} else if p.FetchPartition.Err != nil {
+					pm = pc.newErrorMessage(p.FetchPartition.Err, p.Topic, p.Partition)
+				}
+				_ = pc.handler(pm)
+				eventPool.Put(pm)
+				return
+			}
+
 			for _, record := range p.Records {
 				ctx, sp := pc.htracer.WithProcessSpan(record)
 				ts := time.Now()
@@ -238,6 +280,11 @@ func (pc *consumer) consume() {
 				for _, hdr := range record.Headers {
 					p.msg.Header.Set(hdr.Key, string(hdr.Value))
 				}
+				p.msg.Header.Set("Micro-Offset", strconv.FormatInt(record.Offset, 10))
+				p.msg.Header.Set("Micro-Partition", strconv.FormatInt(int64(record.Partition), 10))
+				p.msg.Header.Set("Micro-Topic", record.Topic)
+				p.msg.Header.Set("Micro-Key", string(record.Key))
+				p.msg.Header.Set("Micro-Timestamp", strconv.FormatInt(record.Timestamp.Unix(), 10))
 				if pc.kopts.Codec.String() == "noop" {
 					p.msg.Body = record.Value
 				} else if pc.opts.BodyOnly {
@@ -263,9 +310,12 @@ func (pc *consumer) consume() {
 							if p.ack {
 								pc.c.MarkCommitRecords(record)
 							} else {
+								if sp != nil {
+									sp.Finish()
+								}
 								eventPool.Put(p)
-								//								pc.connected.Store(0)
-								pc.kopts.Logger.Fatal(pc.kopts.Context, "[kgo] ErrLostMessage wtf?")
+								pm := pc.newErrorMessage(ErrLostMessage, record.Topic, record.Partition)
+								pc.handler(pm)
 								return
 							}
 							eventPool.Put(p)
@@ -274,15 +324,16 @@ func (pc *consumer) consume() {
 							pc.kopts.Meter.Histogram(semconv.SubscribeMessageDurationSeconds, "endpoint", record.Topic, "topic", record.Topic).Update(te.Seconds())
 							continue
 						} else {
-							pc.kopts.Logger.Error(pc.kopts.Context, "[kgo]: unmarshal error", err)
+							pm := pc.newErrorMessage(err, record.Topic, record.Partition)
+							pc.handler(pm)
 						}
 						te := time.Since(ts)
 						pc.kopts.Meter.Counter(semconv.SubscribeMessageInflight, "endpoint", record.Topic, "topic", record.Topic).Dec()
 						pc.kopts.Meter.Summary(semconv.SubscribeMessageLatencyMicroseconds, "endpoint", record.Topic, "topic", record.Topic).Update(te.Seconds())
 						pc.kopts.Meter.Histogram(semconv.SubscribeMessageDurationSeconds, "endpoint", record.Topic, "topic", record.Topic).Update(te.Seconds())
 						eventPool.Put(p)
-						//						pc.connected.Store(0)
-						pc.kopts.Logger.Fatal(pc.kopts.Context, "[kgo] Unmarshal err not handled wtf?")
+						pm := pc.newErrorMessage(ErrLostMessage, record.Topic, record.Partition)
+						pc.handler(pm)
 						if sp != nil {
 							sp.Finish()
 						}
@@ -331,8 +382,8 @@ func (pc *consumer) consume() {
 					pc.c.MarkCommitRecords(record)
 				} else {
 					eventPool.Put(p)
-					//					pc.connected.Store(0)
-					pc.kopts.Logger.Fatal(pc.kopts.Context, "[kgo] ErrLostMessage wtf?")
+					pm := pc.newErrorMessage(ErrLostMessage, record.Topic, record.Partition)
+					pc.handler(pm)
 					if sp != nil {
 						sp.SetStatus(tracer.SpanStatusError, "ErrLostMessage")
 						sp.Finish()
@@ -346,3 +397,113 @@ func (pc *consumer) consume() {
 		}
 	}
 }
+
+func (pc *consumer) newErrorMessage(err error, t string, p int32) *event {
+	pm := eventPool.Get().(*event)
+
+	pm.ack = false
+	pm.msg = &broker.Message{Header: metadata.New(2)}
+	pm.err = err
+	pm.topic = t
+	pm.ctx = context.Background()
+	pm.msg.Header.Set("Micro-Partition", strconv.FormatInt(int64(p), 10))
+	pm.msg.Header.Set("Micro-Topic", t)
+	return pm
+}
+
+func newErrorFetchTopicPartition(err error, t string, p int32) kgo.FetchTopicPartition {
+	return kgo.FetchTopicPartition{
+		Topic: t,
+		FetchPartition: kgo.FetchPartition{
+			Partition: p,
+			Err:       err,
+		},
+	}
+}
+
+var (
+	_ kgo.HookBrokerConnect           = (*Subscriber)(nil)
+	_ kgo.HookBrokerDisconnect        = (*Subscriber)(nil)
+	_ kgo.HookBrokerRead              = (*Subscriber)(nil)
+	_ kgo.HookBrokerWrite             = (*Subscriber)(nil)
+	_ kgo.HookGroupManageError        = (*Subscriber)(nil)
+	_ kgo.HookProduceRecordUnbuffered = (*Subscriber)(nil)
+)
+
+func (s *Subscriber) OnGroupManageError(err error) {
+	if err == nil {
+		return
+	}
+	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)
+		}
+	}
+}
+
+func (s *Subscriber) OnBrokerConnect(_ kgo.BrokerMetadata, _ time.Duration, _ net.Conn, err error) {
+	if err == nil {
+		return
+	}
+	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)
+		}
+	}
+}
+
+func (s *Subscriber) OnBrokerDisconnect(_ kgo.BrokerMetadata, _ net.Conn) {
+}
+
+func (s *Subscriber) OnBrokerWrite(_ kgo.BrokerMetadata, _ int16, _ int, _ time.Duration, _ time.Duration, err error) {
+	if err == nil {
+		return
+	}
+	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)
+		}
+	}
+}
+
+func (s *Subscriber) OnBrokerRead(_ kgo.BrokerMetadata, _ int16, _ int, _ time.Duration, _ time.Duration, err error) {
+	if err == nil {
+		return
+	}
+	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)
+		}
+	}
+}
+
+func (s *Subscriber) OnProduceRecordUnbuffered(_ *kgo.Record, err error) {
+	if err == nil {
+		return
+	}
+	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)
+		}
+	}
+}