unistack-org/micro-broker-kgo
4
0
Fork 2
Code Issues 3 Pull Requests 2 Actions Packages Projects Releases 73 Wiki Activity

new kgo version

Browse Source 
update deps

fixup race conditions

add kfake usage

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
This commit is contained in:
Василий Толстов
2025-05-11 01:49:38 +03:00
parent ccd912adb2
commit 580d5460dd
6 changed files with 104 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
broker.go
View File

@@ -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)
}
}
}

14
go.mod
View File

@@ -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
)

29
go.sum
View File

@@ -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=

4
kgo.go
View File

@@ -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 {

84
kgo_test.go
View File

@@ -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)
}
go func() {
for _, msg := range msgs {
t.Logf("broker publish")
// 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 {

11
subscriber.go
View File

@@ -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))