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

improve performance and correctness

Browse Source 
* properly handle rebalances
* simplify code
* return on NewBroker instance and not interface

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
This commit is contained in:
Василий Толстов
2023-03-13 21:56:44 +03:00
parent 9a69314c25
commit a9b13378f3
7 changed files with 478 additions and 499 deletions
Download Patch File Download Diff File Expand all files Collapse all files

368
kgo.go
View File

@@ -3,26 +3,56 @@ package kgo // import "go.unistack.org/micro-broker-kgo/v3"
import (
"context"
"errors"
"fmt"
"math/rand"
"strings"
"sync"
"time"
"github.com/twmb/franz-go/pkg/kerr"
kgo "github.com/twmb/franz-go/pkg/kgo"
"github.com/twmb/franz-go/pkg/kmsg"
"github.com/twmb/franz-go/pkg/kversion"
"github.com/twmb/franz-go/pkg/kgo"
"go.unistack.org/micro/v3/broker"
"go.unistack.org/micro/v3/logger"
"go.unistack.org/micro/v3/metadata"
"go.unistack.org/micro/v3/util/id"
id "go.unistack.org/micro/v3/util/id"
mrand "go.unistack.org/micro/v3/util/rand"
)
var _ broker.Broker = &kBroker{}
var _ broker.Broker = &Broker{}
type kBroker struct {
writer *kgo.Client // used only to push messages
var ErrLostMessage = errors.New("message not marked for offsets commit and will be lost in next iteration")
var DefaultRetryBackoffFn = func() func(int) time.Duration {
var rngMu sync.Mutex
rng := rand.New(rand.NewSource(time.Now().UnixNano()))
return func(fails int) time.Duration {
const (
min = 100 * time.Millisecond
max = time.Second
)
if fails <= 0 {
return min
}
if fails > 10 {
return max
}
backoff := min * time.Duration(1<<(fails-1))
rngMu.Lock()
jitter := 0.8 + 0.4*rng.Float64()
rngMu.Unlock()
backoff = time.Duration(float64(backoff) * jitter)
if backoff > max {
return max
}
return backoff
}
}()
type Broker struct {
c *kgo.Client
kopts []kgo.Opt
connected bool
init bool
@@ -31,79 +61,34 @@ type kBroker struct {
subs []*subscriber
}
type subscriber struct {
reader *kgo.Client // used only to pull messages
topic string
opts broker.SubscribeOptions
kopts broker.Options
handler broker.Handler
batchhandler broker.BatchHandler
closed bool
done chan struct{}
consumers map[string]map[int32]worker
sync.RWMutex
}
type publication struct {
topic string
err error
sync.RWMutex
msg *broker.Message
ack bool
}
func (p *publication) Topic() string {
return p.topic
}
func (p *publication) Message() *broker.Message {
return p.msg
}
func (p *publication) Ack() error {
p.ack = true
return nil
}
func (p *publication) Error() error {
return p.err
}
func (p *publication) SetError(err error) {
p.err = err
}
func (s *subscriber) Options() broker.SubscribeOptions {
return s.opts
}
func (s *subscriber) Topic() string {
return s.topic
}
func (s *subscriber) Unsubscribe(ctx context.Context) error {
if s.closed {
return nil
}
select {
case <-ctx.Done():
return ctx.Err()
default:
close(s.done)
s.closed = true
}
return nil
}
func (k *kBroker) Address() string {
func (k *Broker) Address() string {
return strings.Join(k.opts.Addrs, ",")
}
func (k *kBroker) Name() string {
func (k *Broker) Name() string {
return k.opts.Name
}
func (k *kBroker) Connect(ctx context.Context) error {
func (k *Broker) connect(ctx context.Context, opts ...kgo.Opt) (*kgo.Client, error) {
var c *kgo.Client
var err error
select {
case <-ctx.Done():
return nil, ctx.Err()
default:
c, err = kgo.NewClient(opts...)
if err == nil {
err = c.Ping(ctx) // check connectivity to cluster
}
if err != nil {
return nil, err
}
}
return c, nil
}
func (k *Broker) Connect(ctx context.Context) error {
k.RLock()
if k.connected {
k.RUnlock()
@@ -116,51 +101,20 @@ func (k *kBroker) Connect(ctx context.Context) error {
nctx = ctx
}
kaddrs := k.opts.Addrs
// shuffle addrs
var rng mrand.Rand
rng.Shuffle(len(kaddrs), func(i, j int) {
kaddrs[i], kaddrs[j] = kaddrs[j], kaddrs[i]
})
kopts := append(k.kopts, kgo.SeedBrokers(kaddrs...))
select {
case <-nctx.Done():
return nctx.Err()
default:
c, err := kgo.NewClient(kopts...)
if err != nil {
return err
}
// Request versions in order to guess Kafka Cluster version
versionsReq := kmsg.NewApiVersionsRequest()
versionsRes, err := versionsReq.RequestWith(ctx, c)
if err != nil {
return fmt.Errorf("failed to request api versions: %w", err)
}
err = kerr.ErrorForCode(versionsRes.ErrorCode)
if err != nil {
return fmt.Errorf("failed to request api versions. Inner kafka error: %w", err)
}
versions := kversion.FromApiVersionsResponse(versionsRes)
if k.opts.Logger.V(logger.InfoLevel) {
logger.Infof(ctx, "[kgo] connected to to kafka cluster version %v", versions.VersionGuess())
}
k.Lock()
k.connected = true
k.writer = c
k.Unlock()
c, err := k.connect(nctx, k.kopts...)
if err != nil {
return err
}
k.Lock()
k.c = c
k.connected = true
k.Unlock()
return nil
}
func (k *kBroker) Disconnect(ctx context.Context) error {
func (k *Broker) Disconnect(ctx context.Context) error {
k.RLock()
if !k.connected {
k.RUnlock()
@@ -168,14 +122,13 @@ func (k *kBroker) Disconnect(ctx context.Context) error {
}
k.RUnlock()
k.Lock()
defer k.Unlock()
nctx := k.opts.Context
if ctx != nil {
nctx = ctx
}
k.Lock()
defer k.Unlock()
select {
case <-nctx.Done():
return nctx.Err()
@@ -185,20 +138,20 @@ func (k *kBroker) Disconnect(ctx context.Context) error {
return err
}
}
k.writer.Close()
if k.c != nil {
k.c.CloseAllowingRebalance()
// k.c.Close()
}
}
k.connected = false
return nil
}
func (k *kBroker) Init(opts ...broker.Option) error {
func (k *Broker) Init(opts ...broker.Option) error {
k.Lock()
defer k.Unlock()
if len(opts) == 0 && k.init {
return nil
}
for _, o := range opts {
o(&k.opts)
}
@@ -222,33 +175,38 @@ func (k *kBroker) Init(opts ...broker.Option) error {
}
}
// kgo.RecordPartitioner(),
k.init = true
return nil
}
func (k *kBroker) Options() broker.Options {
func (k *Broker) Options() broker.Options {
return k.opts
}
func (k *kBroker) BatchPublish(ctx context.Context, msgs []*broker.Message, opts ...broker.PublishOption) error {
func (k *Broker) BatchPublish(ctx context.Context, msgs []*broker.Message, opts ...broker.PublishOption) error {
return k.publish(ctx, msgs, opts...)
}
func (k *kBroker) Publish(ctx context.Context, topic string, msg *broker.Message, opts ...broker.PublishOption) error {
func (k *Broker) Publish(ctx context.Context, topic string, msg *broker.Message, opts ...broker.PublishOption) error {
msg.Header.Set(metadata.HeaderTopic, topic)
return k.publish(ctx, []*broker.Message{msg}, opts...)
}
func (k *kBroker) publish(ctx context.Context, msgs []*broker.Message, opts ...broker.PublishOption) error {
func (k *Broker) publish(ctx context.Context, msgs []*broker.Message, opts ...broker.PublishOption) error {
k.RLock()
if !k.connected {
k.RUnlock()
return broker.ErrNotConnected
k.Lock()
c, err := k.connect(ctx, k.kopts...)
if err != nil {
k.Unlock()
return err
}
k.c = c
k.connected = true
k.Unlock()
}
k.RUnlock()
options := broker.NewPublishOptions(opts...)
records := make([]*kgo.Record, 0, len(msgs))
var errs []string
@@ -264,13 +222,13 @@ func (k *kBroker) publish(ctx context.Context, msgs []*broker.Message, opts ...b
for _, msg := range msgs {
rec := &kgo.Record{Key: key}
rec.Topic, _ = msg.Header.Get(metadata.HeaderTopic)
if k.opts.Codec.String() == "noop" {
if options.BodyOnly {
rec.Value = msg.Body
} else if k.opts.Codec.String() == "noop" {
rec.Value = msg.Body
for k, v := range msg.Header {
rec.Headers = append(rec.Headers, kgo.RecordHeader{Key: k, Value: []byte(v)})
}
} else if options.BodyOnly {
rec.Value = msg.Body
} else {
rec.Value, err = k.opts.Codec.Marshal(msg)
if err != nil {
@@ -280,7 +238,7 @@ func (k *kBroker) publish(ctx context.Context, msgs []*broker.Message, opts ...b
records = append(records, rec)
}
results := k.writer.ProduceSync(ctx, records...)
results := k.c.ProduceSync(ctx, records...)
for _, result := range results {
if result.Err != nil {
errs = append(errs, result.Err.Error())
@@ -294,53 +252,11 @@ func (k *kBroker) publish(ctx context.Context, msgs []*broker.Message, opts ...b
return nil
}
type mlogger struct {
l logger.Logger
ctx context.Context
}
func (l *mlogger) Log(lvl kgo.LogLevel, msg string, args ...interface{}) {
var mlvl logger.Level
switch lvl {
case kgo.LogLevelNone:
return
case kgo.LogLevelError:
mlvl = logger.ErrorLevel
case kgo.LogLevelWarn:
mlvl = logger.WarnLevel
case kgo.LogLevelInfo:
mlvl = logger.InfoLevel
case kgo.LogLevelDebug:
mlvl = logger.DebugLevel
default:
return
}
fields := make(map[string]interface{}, int(len(args)/2))
for i := 0; i < len(args)/2; i += 2 {
fields[fmt.Sprintf("%v", args[i])] = args[i+1]
}
l.l.Fields(fields).Log(l.ctx, mlvl, msg)
}
func (l *mlogger) Level() kgo.LogLevel {
switch l.l.Options().Level {
case logger.ErrorLevel:
return kgo.LogLevelError
case logger.WarnLevel:
return kgo.LogLevelWarn
case logger.InfoLevel:
return kgo.LogLevelInfo
case logger.DebugLevel, logger.TraceLevel:
return kgo.LogLevelDebug
}
return kgo.LogLevelNone
}
func (k *kBroker) BatchSubscribe(ctx context.Context, topic string, handler broker.BatchHandler, opts ...broker.SubscribeOption) (broker.Subscriber, error) {
func (k *Broker) BatchSubscribe(ctx context.Context, topic string, handler broker.BatchHandler, opts ...broker.SubscribeOption) (broker.Subscriber, error) {
return nil, nil
}
func (k *kBroker) Subscribe(ctx context.Context, topic string, handler broker.Handler, opts ...broker.SubscribeOption) (broker.Subscriber, error) {
func (k *Broker) Subscribe(ctx context.Context, topic string, handler broker.Handler, opts ...broker.SubscribeOption) (broker.Subscriber, error) {
options := broker.NewSubscribeOptions(opts...)
if options.Group == "" {
@@ -351,54 +267,47 @@ func (k *kBroker) Subscribe(ctx context.Context, topic string, handler broker.Ha
options.Group = uid
}
kaddrs := k.opts.Addrs
// shuffle addrs
var rng mrand.Rand
rng.Shuffle(len(kaddrs), func(i, j int) {
kaddrs[i], kaddrs[j] = kaddrs[j], kaddrs[i]
})
td := DefaultCommitInterval
commitInterval := DefaultCommitInterval
if k.opts.Context != nil {
if v, ok := k.opts.Context.Value(commitIntervalKey{}).(time.Duration); ok && v > 0 {
td = v
commitInterval = v
}
}
sub := &subscriber{
topic: topic,
done: make(chan struct{}),
opts: options,
handler: handler,
kopts: k.opts,
consumers: make(map[string]map[int32]worker),
consumers: make(map[tp]*consumer),
done: make(chan struct{}),
}
kopts := append(k.kopts,
kgo.SeedBrokers(kaddrs...),
kgo.ConsumerGroup(options.Group),
kgo.ConsumeTopics(topic),
kgo.ConsumeResetOffset(kgo.NewOffset().AtStart()),
kgo.FetchMaxWait(1*time.Second),
// kgo.KeepControlRecords(),
kgo.Balancers(kgo.CooperativeStickyBalancer(), kgo.StickyBalancer()),
kgo.FetchIsolationLevel(kgo.ReadUncommitted()),
kgo.WithHooks(&metrics{meter: k.opts.Meter}),
kgo.AutoCommitMarks(),
kgo.AutoCommitInterval(td),
kgo.AutoCommitInterval(commitInterval),
kgo.OnPartitionsAssigned(sub.assigned),
kgo.OnPartitionsRevoked(sub.revoked),
kgo.OnPartitionsLost(sub.revoked),
kgo.OnPartitionsLost(sub.lost),
kgo.AutoCommitMarks(),
)
reader, err := kgo.NewClient(kopts...)
if options.Context != nil {
if v, ok := options.Context.Value(optionsKey{}).([]kgo.Opt); ok && len(v) > 0 {
kopts = append(kopts, v...)
}
}
c, err := k.connect(ctx, kopts...)
if err != nil {
return nil, err
}
sub.reader = reader
go sub.run(ctx)
sub.c = c
go sub.poll(ctx)
k.Lock()
k.subs = append(k.subs, sub)
@@ -406,45 +315,32 @@ func (k *kBroker) Subscribe(ctx context.Context, topic string, handler broker.Ha
return sub, nil
}
func (k *kBroker) String() string {
func (k *Broker) String() string {
return "kgo"
}
func NewBroker(opts ...broker.Option) *kBroker {
func NewBroker(opts ...broker.Option) *Broker {
rand.Seed(time.Now().Unix())
options := broker.NewOptions(opts...)
if options.Codec.String() != "noop" {
options.Logger.Infof(options.Context, "broker codec not noop, disable plain kafka headers usage")
}
kaddrs := options.Addrs
// shuffle addrs
var rng mrand.Rand
rng.Shuffle(len(kaddrs), func(i, j int) {
kaddrs[i], kaddrs[j] = kaddrs[j], kaddrs[i]
})
kopts := []kgo.Opt{
kgo.DialTimeout(3 * time.Second),
kgo.DisableIdempotentWrite(),
kgo.ProducerBatchCompression(kgo.NoCompression()),
kgo.WithLogger(&mlogger{l: options.Logger, ctx: options.Context}),
kgo.RetryBackoffFn(
func() func(int) time.Duration {
var rng mrand.Rand
return func(fails int) time.Duration {
const (
min = 250 * time.Millisecond
max = 2 * time.Second
)
if fails <= 0 {
return min
}
if fails > 10 {
return max
}
backoff := min * time.Duration(1<<(fails-1))
jitter := 0.8 + 0.4*rng.Float64()
backoff = time.Duration(float64(backoff) * jitter)
if backoff > max {
return max
}
return backoff
}
}(),
),
// kgo.WithLogger(kgo.BasicLogger(os.Stderr, kgo.LogLevelDebug, func() string { return time.Now().Format(time.StampMilli) })),
kgo.WithHooks(&metrics{meter: options.Meter}),
kgo.SeedBrokers(kaddrs...),
kgo.RetryBackoffFn(DefaultRetryBackoffFn),
kgo.BlockRebalanceOnPoll(),
kgo.Balancers(kgo.CooperativeStickyBalancer()),
kgo.FetchIsolationLevel(kgo.ReadUncommitted()),
}
if options.Context != nil {
@@ -453,7 +349,7 @@ func NewBroker(opts ...broker.Option) *kBroker {
}
}
return &kBroker{
return &Broker{
opts: options,
kopts: kopts,
}