pass errors from broker to subscribers

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

.github/workflows/job_sync.yml

@@ -25,7 +25,7 @@ jobs:
         dst_hash=$(git ls-remote ${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY} refs/heads/master | cut -f1)
         echo "src_hash=$src_hash"
         echo "dst_hash=$dst_hash"
-        if [ "$src_hash" != "$dst_hash" ]; then
+        if [ "$src_hash" != "$dst_hash" -a "$src_hash" != "" -a "" != "$dst_hash" ]; then
           echo "sync_needed=true" >> $GITHUB_OUTPUT
         else
           echo "sync_needed=false" >> $GITHUB_OUTPUT

README.md

@@ -1,2 +1,2 @@
 # micro-broker-kgo
-![Coverage](https://img.shields.io/badge/Coverage-66.9%25-yellow)
+![Coverage](https://img.shields.io/badge/Coverage-64.8%25-yellow)

go.mod

@@ -4,18 +4,18 @@ go 1.24.0
 
 require (
 	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 v1.20.2
+	github.com/twmb/franz-go/pkg/kadm v1.17.1
 	github.com/twmb/franz-go/pkg/kfake v0.0.0-20250508175730-72e1646135e3
-	github.com/twmb/franz-go/pkg/kmsg v1.11.2
+	github.com/twmb/franz-go/pkg/kmsg v1.12.0
 	go.opentelemetry.io/otel v1.38.0
-	go.unistack.org/micro/v4 v4.1.21
+	go.unistack.org/micro/v4 v4.1.24
 )
 
 require (
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
 	github.com/google/uuid v1.6.0 // indirect
-	github.com/klauspost/compress v1.18.0 // indirect
+	github.com/klauspost/compress v1.18.1 // indirect
 	github.com/matoous/go-nanoid v1.5.1 // indirect
 	github.com/pierrec/lz4/v4 v4.1.22 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect

go.sum

@@ -8,6 +8,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=
@@ -26,18 +28,26 @@ 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/kfake v0.0.0-20250508175730-72e1646135e3 h1:p24opKWPySAy8xSl8NqRgOv7Q+bX7kdrQirBVRJzQfo=
 github.com/twmb/franz-go/pkg/kfake v0.0.0-20250508175730-72e1646135e3/go.mod h1:7uQs3Ae6HkWT1Y9elMbqtAcNFCI0y6+iS+Phw49L49U=
 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/v4 v4.1.0 h1:qPwL2n/oqh9RE3RTTDgt28XK3QzV597VugQPaw9lKUk=
 go.unistack.org/micro-proto/v4 v4.1.0/go.mod h1:ArmK7o+uFvxSY3dbJhKBBX4Pm1rhWdLEFf3LxBrMtec=
 go.unistack.org/micro/v4 v4.1.21 h1:F9PrbI1BhXSDS0FopwcO5wWrT2Xh38w9VhVZZBHjfg8=
 go.unistack.org/micro/v4 v4.1.21/go.mod h1:nlBXTbx0rQrSZX4HPp2m57PHmpuGPWUd0O+jpUIiPto=
+go.unistack.org/micro/v4 v4.1.24 h1:PbkSWJS3ssB5A0y0tOdOw6u9e2Mk4yombF4yR0Jshvo=
+go.unistack.org/micro/v4 v4.1.24/go.mod h1:nlBXTbx0rQrSZX4HPp2m57PHmpuGPWUd0O+jpUIiPto=
 golang.org/x/crypto v0.43.0 h1:dduJYIi3A3KOfdGOHX8AVZ/jGiyPa3IbBozJ5kNuE04=
 golang.org/x/crypto v0.43.0/go.mod h1:BFbav4mRNlXJL4wNeejLpWxB7wMbc79PdRGhWKncxR0=
 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

@@ -112,6 +112,7 @@ type kgoMessage struct {
 	hdr   metadata.Metadata
 	opts  broker.MessageOptions
 	ack   bool
+	err   error
 }
 
 func (m *kgoMessage) Ack() error {
@@ -135,6 +136,10 @@ func (m *kgoMessage) Topic() string {
 	return ""
 }
 
+func (m *kgoMessage) Error() error {
+	return m.err
+}
+
 func (m *kgoMessage) Unmarshal(dst interface{}, opts ...codec.Option) error {
 	return m.c.Unmarshal(m.body, dst)
 }
@@ -519,15 +524,6 @@ func (b *Broker) fnSubscribe(ctx context.Context, topic string, handler interfac
 
 	options := broker.NewSubscribeOptions(opts...)
 
-	switch handler.(type) {
-	default:
-		return nil, broker.ErrInvalidHandler
-	case func(broker.Message) error:
-		break
-	case func([]broker.Message) error:
-		break
-	}
-
 	if options.Group == "" {
 		uid, err := id.New()
 		if err != nil {
@@ -584,6 +580,7 @@ func (b *Broker) fnSubscribe(ctx context.Context, topic string, handler interfac
 		kgo.OnPartitionsLost(sub.lost),
 		kgo.AutoCommitCallback(sub.autocommit),
 		kgo.AutoCommitMarks(),
+		kgo.WithHooks(sub),
 	)
 
 	if options.Context != nil {

kgo_test.go

@@ -82,7 +82,7 @@ func TestFail(t *testing.T) {
 		for _, msg := range msgs {
 			//		t.Logf("broker publish")
 			if err := b.Publish(ctx, "test.fail", msg); err != nil {
-				t.Fatal(err)
+				// t.Fatal(err)
 			}
 		}
 	}()

meter.go

@@ -14,18 +14,18 @@ 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

subscriber.go

@@ -3,6 +3,8 @@ package kgo
 import (
 	"context"
 	"fmt"
+	"maps"
+	"net"
 	"strconv"
 	"sync"
 	"sync/atomic"
@@ -104,13 +106,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(kgo.ErrClientClosed, t, p)
 				}
 			})
 
@@ -138,7 +151,7 @@ func (s *Subscriber) killConsumers(ctx context.Context, lost map[string][]int32)
 			s.mu.Lock()
 			pc, ok := s.consumers[tps]
 			s.mu.Unlock()
-			if !ok {
+			if !ok || pc == nil {
 				continue
 			}
 			s.mu.Lock()
@@ -154,11 +167,21 @@ 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) {
@@ -178,7 +200,6 @@ 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)
 	}
 }
 
@@ -223,6 +244,27 @@ func (pc *consumer) consume() {
 		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)
+				}
+
+				switch h := pc.handler.(type) {
+				case func(broker.Message) error:
+					_ = h(pm)
+				case func([]broker.Message) error:
+					_ = h([]broker.Message{pm})
+				}
+				if pc.messagePool {
+					messagePool.Put(pm)
+				}
+
+				return
+			}
+
 			for _, record := range p.Records {
 				ctx, sp := pc.htracer.WithProcessSpan(record)
 				ts := time.Now()
@@ -233,19 +275,23 @@ func (pc *consumer) consume() {
 				} else {
 					pm = &kgoMessage{}
 				}
+
 				pm.body = record.Value
 				pm.topic = record.Topic
 				pm.ack = false
 				pm.hdr = metadata.New(len(record.Headers))
 				pm.ctx = ctx
+
 				for _, hdr := range record.Headers {
 					pm.hdr.Set(hdr.Key, string(hdr.Value))
 				}
+
 				pm.hdr.Set("Micro-Offset", strconv.FormatInt(record.Offset, 10))
 				pm.hdr.Set("Micro-Partition", strconv.FormatInt(int64(record.Partition), 10))
 				pm.hdr.Set("Micro-Topic", record.Topic)
 				pm.hdr.Set("Micro-Key", string(record.Key))
 				pm.hdr.Set("Micro-Timestamp", strconv.FormatInt(record.Timestamp.Unix(), 10))
+
 				switch h := pc.handler.(type) {
 				case func(broker.Message) error:
 					err = h(pm)
@@ -269,22 +315,146 @@ func (pc *consumer) consume() {
 
 				ack := pm.ack
 				if pc.messagePool {
-					messagePool.Put(p)
-				}
-				if ack {
-					pc.c.MarkCommitRecords(record)
-				} else {
-					if sp != nil {
-						sp.Finish()
-					}
-					//					pc.connected.Store(0)
-					pc.kopts.Logger.Fatal(pc.kopts.Context, "[kgo] message not commited")
-					return
+					messagePool.Put(pm)
 				}
+
 				if sp != nil {
 					sp.Finish()
 				}
+
+				if ack {
+					pc.c.MarkCommitRecords(record)
+					continue
+				}
+
+				pm := pc.newErrorMessage(ErrLostMessage, p.Topic, p.Partition)
+				switch h := pc.handler.(type) {
+				case func(broker.Message) error:
+					_ = h(pm)
+				case func([]broker.Message) error:
+					_ = h([]broker.Message{pm})
+				}
+				if pc.messagePool {
+					messagePool.Put(pm)
+				}
+				return
 			}
 		}
 	}
 }
+
+func (pc *consumer) newErrorMessage(err error, t string, p int32) *kgoMessage {
+	var pm *kgoMessage
+
+	if pc.messagePool {
+		pm = messagePool.Get().(*kgoMessage)
+	} else {
+		pm = &kgoMessage{}
+	}
+
+	pm.err = err
+	pm.topic = t
+	pm.hdr = metadata.New(2)
+	pm.ctx = context.Background()
+	pm.hdr.Set("Micro-Partition", strconv.FormatInt(int64(p), 10))
+	pm.hdr.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)
+		}
+	}
+}