diff --git a/go.mod b/go.mod index 543740d..efc12ac 100644 --- a/go.mod +++ b/go.mod @@ -1,17 +1,17 @@ module go.unistack.org/micro-broker-kgo/v3 -go 1.23.8 +go 1.24.0 toolchain go1.24.3 require ( github.com/google/uuid v1.6.0 - github.com/stretchr/testify v1.10.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.0 + github.com/twmb/franz-go/pkg/kadm v1.16.1 github.com/twmb/franz-go/pkg/kmsg v1.11.2 - go.opentelemetry.io/otel v1.34.0 - go.unistack.org/micro/v3 v3.11.44 + go.opentelemetry.io/otel v1.38.0 + go.unistack.org/micro/v3 v3.11.45 ) require ( @@ -23,10 +23,10 @@ require ( github.com/pierrec/lz4/v4 v4.1.22 // indirect github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect go.unistack.org/micro-proto/v3 v3.4.1 // indirect - golang.org/x/crypto v0.38.0 // indirect - golang.org/x/sys v0.33.0 // indirect - google.golang.org/genproto/googleapis/rpc v0.0.0-20250224174004-546df14abb99 // indirect - google.golang.org/grpc v1.70.0 // indirect - google.golang.org/protobuf v1.36.5 // 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/grpc v1.76.0 // indirect + google.golang.org/protobuf v1.36.10 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect ) diff --git a/go.sum b/go.sum index 7105789..b62e0a4 100644 --- a/go.sum +++ b/go.sum @@ -7,8 +7,8 @@ github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= -github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI= -github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= +github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= +github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= 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= @@ -27,34 +27,34 @@ github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/fJaraNFVN+nFs= github.com/rogpeppe/go-internal v1.13.1 h1:KvO1DLK/DRN07sQ1LQKScxyZJuNnedQ5/wKSR38lUII= github.com/rogpeppe/go-internal v1.13.1/go.mod h1:uMEvuHeurkdAXX61udpOXGD/AzZDWNMNyH2VO9fmH0o= -github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA= -github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= +github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= +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/pkg/kadm v1.16.0 h1:STMs1t5lYR5mR974PSiwNzE5TvsosByTp+rKXLOhAjE= -github.com/twmb/franz-go/pkg/kadm v1.16.0/go.mod h1:MUdcUtnf9ph4SFBLLA/XxE29rvLhWYLM9Ygb8dfSCvw= +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/kmsg v1.11.2 h1:hIw75FpwcAjgeyfIGFqivAvwC5uNIOWRGvQgZhH4mhg= github.com/twmb/franz-go/pkg/kmsg v1.11.2/go.mod h1:CFfkkLysDNmukPYhGzuUcDtf46gQSqCZHMW1T4Z+wDE= -go.opentelemetry.io/otel v1.34.0 h1:zRLXxLCgL1WyKsPVrgbSdMN4c0FMkDAskSTQP+0hdUY= -go.opentelemetry.io/otel v1.34.0/go.mod h1:OWFPOQ+h4G8xpyjgqo4SxJYdDQ/qmRH+wivy7zzx9oI= +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.44 h1:A+T8zVcL2vlL66kn/Y4rqhtBybLO829wFEYZJYorDOU= -go.unistack.org/micro/v3 v3.11.44/go.mod h1:13EFW2ps3BN9mpYbp9K0oQu/VDjEN6LJ4wwdom7hcXQ= -golang.org/x/crypto v0.38.0 h1:jt+WWG8IZlBnVbomuhg2Mdq0+BBQaHbtqHEFEigjUV8= -golang.org/x/crypto v0.38.0/go.mod h1:MvrbAqul58NNYPKnOra203SB9vpuZW0e+RRZV+Ggqjw= -golang.org/x/net v0.33.0 h1:74SYHlV8BIgHIFC/LrYkOGIwL19eTYXQ5wc6TBuO36I= -golang.org/x/net v0.33.0/go.mod h1:HXLR5J+9DxmrqMwG9qjGCxZ+zKXxBru04zlTvWlWuN4= -golang.org/x/sys v0.33.0 h1:q3i8TbbEz+JRD9ywIRlyRAQbM0qF7hu24q3teo2hbuw= -golang.org/x/sys v0.33.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k= -golang.org/x/text v0.25.0 h1:qVyWApTSYLk/drJRO5mDlNYskwQznZmkpV2c8q9zls4= -golang.org/x/text v0.25.0/go.mod h1:WEdwpYrmk1qmdHvhkSTNPm3app7v4rsT8F2UD6+VHIA= -google.golang.org/genproto/googleapis/rpc v0.0.0-20250224174004-546df14abb99 h1:ZSlhAUqC4r8TPzqLXQ0m3upBNZeF+Y8jQ3c4CR3Ujms= -google.golang.org/genproto/googleapis/rpc v0.0.0-20250224174004-546df14abb99/go.mod h1:LuRYeWDFV6WOn90g357N17oMCaxpgCnbi/44qJvDn2I= -google.golang.org/grpc v1.70.0 h1:pWFv03aZoHzlRKHWicjsZytKAiYCtNS0dHbXnIdq7jQ= -google.golang.org/grpc v1.70.0/go.mod h1:ofIJqVKDXx/JiXrwr2IG4/zwdH9txy3IlF40RmcJSQw= -google.golang.org/protobuf v1.36.5 h1:tPhr+woSbjfYvY6/GPufUoYizxw1cF/yFoxJ2fmpwlM= -google.golang.org/protobuf v1.36.5/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE= +go.unistack.org/micro/v3 v3.11.45 h1:fjTLZYWgsVf9FIMZBxOg8ios2/tmyimnjZrsrxEUeXU= +go.unistack.org/micro/v3 v3.11.45/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= +golang.org/x/net v0.45.0/go.mod h1:ECOoLqd5U3Lhyeyo/QDCEVQ4sNgYsqvCZ722XogGieY= +golang.org/x/sys v0.37.0 h1:fdNQudmxPjkdUTPnLn5mdQv7Zwvbvpaxqs831goi9kQ= +golang.org/x/sys v0.37.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= +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/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= +google.golang.org/protobuf v1.36.10/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= diff --git a/kgo.go b/kgo.go index 6ad858f..bd72dea 100644 --- a/kgo.go +++ b/kgo.go @@ -6,12 +6,14 @@ import ( "errors" "fmt" "math/rand/v2" + "strconv" "strings" "sync" "sync/atomic" "time" "github.com/google/uuid" + "github.com/twmb/franz-go/pkg/kadm" "github.com/twmb/franz-go/pkg/kgo" "github.com/twmb/franz-go/pkg/kmsg" "go.unistack.org/micro/v3/broker" @@ -66,7 +68,7 @@ type Broker struct { subs []*Subscriber opts broker.Options - sync.RWMutex + mu sync.RWMutex init bool } @@ -184,10 +186,72 @@ func (k *Broker) Connect(ctx context.Context) error { return err } - k.Lock() + k.mu.Lock() k.c = c k.connected.Store(1) - k.Unlock() + k.mu.Unlock() + + exposeLag := false + if k.opts.Context != nil { + if v, ok := k.opts.Context.Value(exposeLagKey{}).(bool); ok && v { + exposeLag = v + } + } + + if exposeLag { + var mu sync.Mutex + var lastUpdate time.Time + type pl struct { + p string + l float64 + } + + lag := make(map[string]map[string]pl) // topic => group => partition => lag + ac := kadm.NewClient(k.c) + + updateStats := func() { + mu.Lock() + if time.Since(lastUpdate) < DefaultStatsInterval { + return + } + mu.Unlock() + + k.mu.Lock() + groups := make([]string, 0, len(k.subs)) + for _, g := range k.subs { + groups = append(groups, g.opts.Group) + } + k.mu.Unlock() + + dgls, err := ac.Lag(ctx, groups...) + if err != nil || !dgls.Ok() { + k.opts.Logger.Error(k.opts.Context, "kgo describe group lag error", err) + return + } + + for gn, dgl := range dgls { + for tn, lmap := range dgl.Lag { + if _, ok := lag[tn]; !ok { + lag[tn] = make(map[string]pl) + } + for p, l := range lmap { + lag[tn][gn] = pl{p: strconv.Itoa(int(p)), l: float64(l.Lag)} + } + } + } + } + + for tn, dg := range lag { + for gn, gl := range dg { + k.opts.Meter.Gauge(semconv.BrokerGroupLag, + func() float64 { updateStats(); return gl.l }, + "topic", tn, + "group", gn, + "partition", gl.p) + } + } + + } return nil } @@ -205,8 +269,8 @@ func (k *Broker) Disconnect(ctx context.Context) error { ctx, span = k.opts.Tracer.Start(ctx, "Disconnect") defer span.Finish() - k.Lock() - defer k.Unlock() + k.mu.Lock() + defer k.mu.Unlock() select { case <-nctx.Done(): return nctx.Err() @@ -231,8 +295,8 @@ func (k *Broker) Disconnect(ctx context.Context) error { } func (k *Broker) Init(opts ...broker.Option) error { - k.Lock() - defer k.Unlock() + k.mu.Lock() + defer k.mu.Unlock() if len(opts) == 0 && k.init { return nil @@ -455,9 +519,9 @@ func (k *Broker) Subscribe(ctx context.Context, topic string, handler broker.Han go sub.poll(ctx) - k.Lock() + k.mu.Lock() k.subs = append(k.subs, sub) - k.Unlock() + k.mu.Unlock() return sub, nil } diff --git a/options.go b/options.go index 8f69673..1154a63 100644 --- a/options.go +++ b/options.go @@ -15,7 +15,7 @@ var ( DefaultCommitInterval = 5 * time.Second // DefaultStatsInterval specifies how fast check consumer lag - DefaultStatsInterval = 5 * time.Second + DefaultStatsInterval = 30 * time.Second // DefaultSubscribeMaxInflight specifies how much messages keep inflight DefaultSubscribeMaxInflight = 100 @@ -120,3 +120,10 @@ func PublishPromise(fn func(*kgo.Record, error)) broker.PublishOption { func ClientPublishPromise(fn func(*kgo.Record, error)) client.PublishOption { return client.SetPublishOption(publishPromiseKey{}, fn) } + +type exposeLagKey struct{} + +// ExposeLag enabled subscriber lag via [meter.Meter] +func ExposeLag(b bool) broker.Option { + return broker.SetOption(exposeLagKey{}, b) +} diff --git a/subscriber.go b/subscriber.go index e0066cd..7027bfe 100644 --- a/subscriber.go +++ b/subscriber.go @@ -3,12 +3,10 @@ package kgo import ( "context" "fmt" - "strconv" "sync" "sync/atomic" "time" - "github.com/twmb/franz-go/pkg/kadm" "github.com/twmb/franz-go/pkg/kgo" "github.com/twmb/franz-go/pkg/kmsg" "go.unistack.org/micro/v3/broker" @@ -97,46 +95,13 @@ func (s *Subscriber) Unsubscribe(ctx context.Context) error { func (s *Subscriber) poll(ctx context.Context) { maxInflight := DefaultSubscribeMaxInflight + if s.opts.Context != nil { if n, ok := s.opts.Context.Value(subscribeMaxInflightKey{}).(int); n > 0 && ok { maxInflight = n } } - go func() { - ac := kadm.NewClient(s.c) - ticker := time.NewTicker(DefaultStatsInterval) - - for { - select { - case <-ctx.Done(): - ticker.Stop() - return - case <-ticker.C: - dgls, err := ac.Lag(ctx, s.opts.Group) - if err != nil || !dgls.Ok() { - continue - } - - dgl, ok := dgls[s.opts.Group] - if !ok { - continue - } - lmap, ok := dgl.Lag[s.topic] - if !ok { - continue - } - - s.Lock() - for p, l := range lmap { - s.kopts.Meter.Counter(semconv.BrokerGroupLag, "topic", s.topic, "group", s.opts.Group, "partition", strconv.Itoa(int(p))).Set(uint64(l.Lag)) - } - s.Unlock() - - } - } - }() - for { select { case <-ctx.Done(): @@ -151,7 +116,11 @@ func (s *Subscriber) poll(ctx context.Context) { return } fetches.EachError(func(t string, p int32, err error) { - s.kopts.Logger.Fatal(ctx, fmt.Sprintf("[kgo] fetch topic %s partition %d error", t, p), err) + 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) + } }) fetches.EachPartition(func(p kgo.FetchTopicPartition) {