From e34f57a51593e58dd29a43b2742b27ebf75f67a7 Mon Sep 17 00:00:00 2001 From: Vasiliy Tolstov Date: Sun, 11 May 2025 01:49:38 +0300 Subject: [PATCH] new kgo version update deps fixup race conditions add kfake usage Signed-off-by: Vasiliy Tolstov --- broker.go | 26 ++++++++------ go.mod | 14 ++++---- go.sum | 29 ++++++++-------- kgo.go | 4 +++ kgo_test.go | 94 +++++++++++++++++++++++++++++---------------------- subscriber.go | 11 +++++- 6 files changed, 104 insertions(+), 74 deletions(-) diff --git a/broker.go b/broker.go index fc0d129..8b3714e 100644 --- a/broker.go +++ b/broker.go @@ -26,8 +26,8 @@ var ( ) func (m *hookEvent) OnGroupManageError(err error) { - if err != nil { - // m.connected.Store(0) + if err != nil && !kgo.IsRetryableBrokerErr(err) { + m.connected.Store(0) if m.fatalOnError { m.log.Fatal(context.TODO(), "kgo.OnGroupManageError", err) } @@ -35,8 +35,8 @@ func (m *hookEvent) OnGroupManageError(err error) { } func (m *hookEvent) OnBrokerConnect(_ kgo.BrokerMetadata, _ time.Duration, _ net.Conn, err error) { - if err != nil { - // m.connected.Store(0) + if err != nil && !kgo.IsRetryableBrokerErr(err) { + m.connected.Store(0) if m.fatalOnError { m.log.Fatal(context.TODO(), "kgo.OnBrokerConnect", err) } @@ -48,8 +48,8 @@ func (m *hookEvent) OnBrokerDisconnect(_ kgo.BrokerMetadata, _ net.Conn) { } func (m *hookEvent) OnBrokerWrite(_ kgo.BrokerMetadata, _ int16, _ int, _ time.Duration, _ time.Duration, err error) { - if err != nil { - // m.connected.Store(0) + if err != nil && !kgo.IsRetryableBrokerErr(err) { + m.connected.Store(0) if m.fatalOnError { m.log.Fatal(context.TODO(), "kgo.OnBrokerWrite", err) } @@ -57,13 +57,19 @@ func (m *hookEvent) OnBrokerWrite(_ kgo.BrokerMetadata, _ int16, _ int, _ time.D } func (m *hookEvent) OnBrokerRead(_ kgo.BrokerMetadata, _ int16, _ int, _ time.Duration, _ time.Duration, err error) { - if err != nil { - // m.connected.Store(0) + if err != nil && !kgo.IsRetryableBrokerErr(err) { + m.connected.Store(0) + if m.fatalOnError { + m.log.Fatal(context.TODO(), "kgo.OnBrokerRead", err) + } } } func (m *hookEvent) OnProduceRecordUnbuffered(_ *kgo.Record, err error) { - if err != nil { - // m.connected.Store(0) + if err != nil && !kgo.IsRetryableBrokerErr(err) { + m.connected.Store(0) + if m.fatalOnError { + m.log.Fatal(context.TODO(), "kgo.OnProduceRecordUnbuffered", err) + } } } diff --git a/go.mod b/go.mod index c209bd2..8d56889 100644 --- a/go.mod +++ b/go.mod @@ -1,24 +1,24 @@ module go.unistack.org/micro-broker-kgo/v4 -go 1.23.0 +go 1.23.8 require ( - github.com/twmb/franz-go v1.18.1 + github.com/twmb/franz-go v1.19.1 github.com/twmb/franz-go/pkg/kadm v1.16.0 + github.com/twmb/franz-go/pkg/kfake v0.0.0-20250508175730-72e1646135e3 github.com/twmb/franz-go/pkg/kmsg v1.11.2 go.opentelemetry.io/otel v1.35.0 - go.unistack.org/micro/v4 v4.1.7 + go.unistack.org/micro-codec-json/v4 v4.1.0 + go.unistack.org/micro/v4 v4.1.13 ) require ( - github.com/ash3in/uuidv8 v1.2.0 // indirect github.com/google/uuid v1.6.0 // indirect github.com/klauspost/compress v1.18.0 // indirect github.com/matoous/go-nanoid v1.5.1 // indirect github.com/pierrec/lz4/v4 v4.1.22 // indirect - github.com/spf13/cast v1.7.1 // indirect + github.com/spf13/cast v1.8.0 // indirect go.unistack.org/micro-proto/v4 v4.1.0 // indirect - golang.org/x/crypto v0.37.0 // indirect + golang.org/x/crypto v0.38.0 // indirect google.golang.org/protobuf v1.36.6 // indirect - gopkg.in/yaml.v3 v3.0.1 // indirect ) diff --git a/go.sum b/go.sum index 7ca7ce9..697ddab 100644 --- a/go.sum +++ b/go.sum @@ -1,7 +1,3 @@ -github.com/DATA-DOG/go-sqlmock v1.5.2 h1:OcvFkGmslmlZibjAjaHm3L//6LiuBgolP7OputlJIzU= -github.com/DATA-DOG/go-sqlmock v1.5.2/go.mod h1:88MAG/4G7SMwSE3CeA0ZKzrT5CiOU3OJ+JlNzwDqpNU= -github.com/ash3in/uuidv8 v1.2.0 h1:2oogGdtCPwaVtyvPPGin4TfZLtOGE5F+W++E880G6SI= -github.com/ash3in/uuidv8 v1.2.0/go.mod h1:BnU0wJBxnzdEKmVg4xckBkD+VZuecTFTUP3M0dWgyY4= github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM= github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/frankban/quicktest v1.14.6 h1:7Xjx+VpznH+oBnejlPUj8oUpdxnVs4f8XU8WnHkI4W8= @@ -24,28 +20,31 @@ github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRI github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= 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/spf13/cast v1.7.1 h1:cuNEagBQEHWN1FnbGEjCXL2szYEXqfJPbP2HNUaca9Y= -github.com/spf13/cast v1.7.1/go.mod h1:ancEpBxwJDODSW/UG4rDrAqiKolqNNh2DX3mk86cAdo= +github.com/spf13/cast v1.8.0 h1:gEN9K4b8Xws4EX0+a0reLmhq8moKn7ntRlQYgjPeCDk= +github.com/spf13/cast v1.8.0/go.mod h1:ancEpBxwJDODSW/UG4rDrAqiKolqNNh2DX3mk86cAdo= 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/twmb/franz-go v1.18.1 h1:D75xxCDyvTqBSiImFx2lkPduE39jz1vaD7+FNc+vMkc= -github.com/twmb/franz-go v1.18.1/go.mod h1:Uzo77TarcLTUZeLuGq+9lNpSkfZI+JErv7YJhlDjs9M= +github.com/twmb/franz-go v1.19.0 h1:FzBAPUeaip68X9cbLDesgQesa5zxKVaZMk+du98vj3c= +github.com/twmb/franz-go v1.19.0/go.mod h1:4kFJ5tmbbl7asgwAGVuyG1ZMx0NNpYk7EqflvWfPCpM= +github.com/twmb/franz-go v1.19.1 h1:cOhDFUkGvUFHSQ7UYW6bO77BJa2fYEk5mA2AX+1NIdE= +github.com/twmb/franz-go v1.19.1/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/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= go.opentelemetry.io/otel v1.35.0 h1:xKWKPxrxB6OtMCbmMY021CqC45J+3Onta9MqjhnusiQ= go.opentelemetry.io/otel v1.35.0/go.mod h1:UEqy8Zp11hpkUrL73gSlELM0DupHoiq72dR+Zqel/+Y= +go.unistack.org/micro-codec-json/v4 v4.1.0 h1:iydeSkt3ee7IPU0dHHKlGN97lw+YFQasBk9rdv0woYA= +go.unistack.org/micro-codec-json/v4 v4.1.0/go.mod h1:aUg86elSlURSynTAetDAAXj/VzFDwwcg92QNrRzcvrM= 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.7 h1:4/dSEMTqxYoHimn/8wKDohfTXi2zDy6eWXYBCnBaZdc= -go.unistack.org/micro/v4 v4.1.7/go.mod h1:lr3oYED8Ay1vjK68QqRw30QOtdk/ffpZqMFDasOUhKw= -golang.org/x/crypto v0.37.0 h1:kJNSjF/Xp7kU0iB2Z+9viTPMW4EqqsrywMXLJOOsXSE= -golang.org/x/crypto v0.37.0/go.mod h1:vg+k43peMZ0pUMhYmVAWysMK35e6ioLh3wB8ZCAfbVc= +go.unistack.org/micro/v4 v4.1.13 h1:1IEQwiIwHdypZN4dWmsWN83Plq9bdkQ4U8aYZsT17s8= +go.unistack.org/micro/v4 v4.1.13/go.mod h1:xleO2M5Yxh4s6I+RUcLrEpUjobefh+71ctrdIfn7TUs= +golang.org/x/crypto v0.38.0 h1:jt+WWG8IZlBnVbomuhg2Mdq0+BBQaHbtqHEFEigjUV8= +golang.org/x/crypto v0.38.0/go.mod h1:MvrbAqul58NNYPKnOra203SB9vpuZW0e+RRZV+Ggqjw= google.golang.org/protobuf v1.36.6 h1:z1NpPI8ku2WgiWnf+t9wTPsn6eP1L7ksHUlkfLvd9xY= google.golang.org/protobuf v1.36.6/go.mod h1:jduwjTPXsFjZGTmRluh+L6NjiWu7pchiJ2/5YcXBHnY= -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= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/kgo.go b/kgo.go index 3904508..42690d2 100644 --- a/kgo.go +++ b/kgo.go @@ -152,6 +152,10 @@ func (b *Broker) newCodec(ct string) (codec.Codec, error) { func (b *Broker) NewMessage(ctx context.Context, hdr metadata.Metadata, body interface{}, opts ...broker.PublishOption) (broker.Message, error) { options := broker.NewPublishOptions(opts...) + if options.ContentType == "" { + options.ContentType = b.opts.ContentType + } + m := &kgoMessage{ctx: ctx, hdr: hdr, opts: options} c, err := b.newCodec(m.opts.ContentType) if err == nil { diff --git a/kgo_test.go b/kgo_test.go index 0399a4b..fd549d7 100644 --- a/kgo_test.go +++ b/kgo_test.go @@ -2,15 +2,15 @@ package kgo_test import ( "context" - "os" - "strings" "sync/atomic" "testing" "time" + "github.com/twmb/franz-go/pkg/kfake" kg "github.com/twmb/franz-go/pkg/kgo" kgo "go.unistack.org/micro-broker-kgo/v4" "go.unistack.org/micro/v4/broker" + "go.unistack.org/micro/v4/codec" "go.unistack.org/micro/v4/logger" "go.unistack.org/micro/v4/logger/slog" "go.unistack.org/micro/v4/metadata" @@ -20,31 +20,33 @@ var ( msgcnt = int64(1200) group = "38" prefill = true - loglevel = logger.DebugLevel + loglevel = logger.ErrorLevel + cluster *kfake.Cluster ) -func TestFail(t *testing.T) { - if tr := os.Getenv("INTEGRATION_TESTS"); len(tr) > 0 { - t.Skip() - } +func TestMain(m *testing.M) { + cluster = kfake.MustCluster( + kfake.AllowAutoTopicCreation(), + ) + defer cluster.Close() + m.Run() +} +func TestFail(t *testing.T) { logger.DefaultLogger = slog.NewLogger() if err := logger.DefaultLogger.Init(logger.WithLevel(loglevel)); err != nil { t.Fatal(err) } ctx := context.Background() - var addrs []string - if addr := os.Getenv("BROKER_ADDRS"); len(addr) == 0 { - addrs = []string{"127.0.0.1:9092"} - } else { - addrs = strings.Split(addr, ",") - } - b := kgo.NewBroker( - broker.Addrs(addrs...), + broker.ContentType("application/octet-stream"), + broker.Codec("application/octet-stream", codec.NewCodec()), + broker.Addrs(cluster.ListenAddrs()...), kgo.CommitInterval(5*time.Second), - kgo.Options(kg.ClientID("test"), kg.FetchMaxBytes(10*1024*1024), + kgo.Options( + kg.ClientID("test"), + kg.FetchMaxBytes(10*1024*1024), kg.AllowAutoTopicCreation(), ), ) @@ -69,23 +71,28 @@ func TestFail(t *testing.T) { t.Logf("broker health %v", b.Health()) msgs := make([]broker.Message, 0, msgcnt) for i := int64(0); i < msgcnt; i++ { - m, _ := b.NewMessage(ctx, metadata.Pairs("hkey", "hval"), []byte(`test`)) + m, err := b.NewMessage(ctx, metadata.Pairs("hkey", "hval"), []byte(`test`)) + if err != nil { + t.Fatal(err) + } msgs = append(msgs, m) } - for _, msg := range msgs { - t.Logf("broker publish") - if err := b.Publish(ctx, "test", msg); err != nil { - t.Fatal(err) + go func() { + for _, msg := range msgs { + // t.Logf("broker publish") + if err := b.Publish(ctx, "test", msg); err != nil { + t.Fatal(err) + } } - } + }() // t.Skip() idx := int64(0) fn := func(msg broker.Message) error { atomic.AddInt64(&idx, 1) - time.Sleep(500 * time.Millisecond) - t.Logf("ack") + time.Sleep(100 * time.Millisecond) + // t.Logf("ack") return msg.Ack() } @@ -96,6 +103,7 @@ func TestFail(t *testing.T) { if err != nil { t.Fatal(err) } + defer func() { if err := sub.Unsubscribe(ctx); err != nil { t.Fatal(err) @@ -108,17 +116,26 @@ func TestFail(t *testing.T) { t.Logf("health works") break } - time.Sleep(1 * time.Second) + t.Logf("health sleep") + time.Sleep(100 * time.Millisecond) + if err := b.Disconnect(ctx); err != nil { + t.Fatal(err) + } } } func TestConnect(t *testing.T) { - var addrs []string ctx := context.TODO() b := kgo.NewBroker( - broker.Addrs(addrs...), + broker.ContentType("application/octet-stream"), + broker.Codec("application/octet-stream", codec.NewCodec()), + broker.Addrs(cluster.ListenAddrs()...), kgo.CommitInterval(5*time.Second), - kgo.Options(kg.ClientID("test"), kg.FetchMaxBytes(10*1024*1024)), + kgo.Options( + kg.ClientID("test"), + kg.FetchMaxBytes(10*1024*1024), + kg.AllowAutoTopicCreation(), + ), ) if err := b.Init(); err != nil { t.Fatal(err) @@ -130,26 +147,21 @@ func TestConnect(t *testing.T) { } func TestPubSub(t *testing.T) { - if tr := os.Getenv("INTEGRATION_TESTS"); len(tr) > 0 { - t.Skip() - } - if err := logger.DefaultLogger.Init(logger.WithLevel(loglevel)); err != nil { t.Fatal(err) } ctx := context.Background() - var addrs []string - if addr := os.Getenv("BROKER_ADDRS"); len(addr) == 0 { - addrs = []string{"127.0.0.1:29091", "127.0.0.2:29092", "127.0.0.3:29093"} - } else { - addrs = strings.Split(addr, ",") - } - b := kgo.NewBroker( - broker.Addrs(addrs...), + broker.ContentType("application/octet-stream"), + broker.Codec("application/octet-stream", codec.NewCodec()), + broker.Addrs(cluster.ListenAddrs()...), kgo.CommitInterval(5*time.Second), - kgo.Options(kg.ClientID("test"), kg.FetchMaxBytes(10*1024*1024)), + kgo.Options( + kg.ClientID("test"), + kg.FetchMaxBytes(10*1024*1024), + kg.AllowAutoTopicCreation(), + ), ) if err := b.Init(); err != nil { diff --git a/subscriber.go b/subscriber.go index d23feb5..5069a80 100644 --- a/subscriber.go +++ b/subscriber.go @@ -146,7 +146,12 @@ func (s *Subscriber) poll(ctx context.Context) { fetches.EachPartition(func(p kgo.FetchTopicPartition) { tps := tp{p.Topic, p.Partition} - s.consumers[tps].recs <- p + s.Lock() + c := s.consumers[tps] + s.Unlock() + if c != nil { + c.recs <- p + } }) s.c.AllowRebalance() } @@ -160,11 +165,15 @@ func (s *Subscriber) killConsumers(ctx context.Context, lost map[string][]int32) for topic, partitions := range lost { for _, partition := range partitions { tps := tp{topic, partition} + s.Lock() pc, ok := s.consumers[tps] + s.Unlock() if !ok { continue } + s.Lock() delete(s.consumers, tps) + s.Unlock() close(pc.quit) if s.kopts.Logger.V(logger.DebugLevel) { s.kopts.Logger.Debug(ctx, fmt.Sprintf("[kgo] waiting for work to finish topic %s partition %d", topic, partition))