From cced3b84c11c84a54f8e6602f3fa378542b2589b Mon Sep 17 00:00:00 2001 From: pugnack Date: Thu, 10 Jul 2025 14:51:26 +0500 Subject: [PATCH] remove noisy logs from event hooks and add tests (#193) --- errors.go | 10 + go.mod | 4 + go.sum | 3 + broker.go => hook_event.go | 69 ++++-- hook_event_test.go | 471 +++++++++++++++++++++++++++++++++++++ 5 files changed, 539 insertions(+), 18 deletions(-) create mode 100644 errors.go rename broker.go => hook_event.go (50%) create mode 100644 hook_event_test.go diff --git a/errors.go b/errors.go new file mode 100644 index 0000000..ef173af --- /dev/null +++ b/errors.go @@ -0,0 +1,10 @@ +package kgo + +import ( + "context" + "errors" +) + +func isContextError(err error) bool { + return errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) +} diff --git a/go.mod b/go.mod index 3269540..858b071 100644 --- a/go.mod +++ b/go.mod @@ -3,6 +3,7 @@ module go.unistack.org/micro-broker-kgo/v4 go 1.23.8 require ( + github.com/stretchr/testify v1.10.0 github.com/twmb/franz-go v1.19.5 github.com/twmb/franz-go/pkg/kadm v1.16.0 github.com/twmb/franz-go/pkg/kfake v0.0.0-20250508175730-72e1646135e3 @@ -12,12 +13,15 @@ require ( ) 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/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 github.com/spf13/cast v1.9.2 // indirect go.unistack.org/micro-proto/v4 v4.1.0 // indirect golang.org/x/crypto v0.39.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 169e384..999e454 100644 --- a/go.sum +++ b/go.sum @@ -42,5 +42,8 @@ golang.org/x/crypto v0.39.0 h1:SHs+kF4LP+f+p14esP5jAoDpHU8Gu/v9lFRK6IT5imM= golang.org/x/crypto v0.39.0/go.mod h1:L+Xg3Wf6HoL4Bn4238Z6ft6KfEpN0tJGo53AAPC632U= 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/broker.go b/hook_event.go similarity index 50% rename from broker.go rename to hook_event.go index 8b3714e..83afda9 100644 --- a/broker.go +++ b/hook_event.go @@ -26,50 +26,83 @@ var ( ) func (m *hookEvent) OnGroupManageError(err error) { - if err != nil && !kgo.IsRetryableBrokerErr(err) { - m.connected.Store(0) + switch { + case err == nil || isContextError(err) || kgo.IsRetryableBrokerErr(err): + return + default: + ctx := context.TODO() + logMsg := "kgo.OnGroupManageError" + if m.fatalOnError { - m.log.Fatal(context.TODO(), "kgo.OnGroupManageError", err) + m.log.Fatal(ctx, logMsg, err) + } else { + m.log.Error(ctx, logMsg, err) } } } func (m *hookEvent) OnBrokerConnect(_ kgo.BrokerMetadata, _ time.Duration, _ net.Conn, err error) { - if err != nil && !kgo.IsRetryableBrokerErr(err) { - m.connected.Store(0) + switch { + case err == nil || isContextError(err) || kgo.IsRetryableBrokerErr(err): + return + default: + ctx := context.TODO() + logMsg := "kgo.OnBrokerConnect" + if m.fatalOnError { - m.log.Fatal(context.TODO(), "kgo.OnBrokerConnect", err) + m.log.Fatal(ctx, logMsg, err) + } else { + m.log.Error(ctx, logMsg, err) } } } -func (m *hookEvent) OnBrokerDisconnect(_ kgo.BrokerMetadata, _ net.Conn) { - // m.connected.Store(0) -} +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 && !kgo.IsRetryableBrokerErr(err) { - m.connected.Store(0) + switch { + case err == nil || isContextError(err) || kgo.IsRetryableBrokerErr(err): + return + default: + ctx := context.TODO() + logMsg := "kgo.OnBrokerWrite" + if m.fatalOnError { - m.log.Fatal(context.TODO(), "kgo.OnBrokerWrite", err) + m.log.Fatal(ctx, logMsg, err) + } else { + m.log.Error(ctx, logMsg, err) } } } func (m *hookEvent) OnBrokerRead(_ kgo.BrokerMetadata, _ int16, _ int, _ time.Duration, _ time.Duration, err error) { - if err != nil && !kgo.IsRetryableBrokerErr(err) { - m.connected.Store(0) + switch { + case err == nil || isContextError(err) || kgo.IsRetryableBrokerErr(err): + return + default: + ctx := context.TODO() + logMsg := "kgo.OnBrokerRead" + if m.fatalOnError { - m.log.Fatal(context.TODO(), "kgo.OnBrokerRead", err) + m.log.Fatal(ctx, logMsg, err) + } else { + m.log.Error(ctx, logMsg, err) } } } func (m *hookEvent) OnProduceRecordUnbuffered(_ *kgo.Record, err error) { - if err != nil && !kgo.IsRetryableBrokerErr(err) { - m.connected.Store(0) + switch { + case err == nil || isContextError(err) || kgo.IsRetryableBrokerErr(err): + return + default: + ctx := context.TODO() + logMsg := "kgo.OnProduceRecordUnbuffered" + if m.fatalOnError { - m.log.Fatal(context.TODO(), "kgo.OnProduceRecordUnbuffered", err) + m.log.Fatal(ctx, logMsg, err) + } else { + m.log.Error(ctx, logMsg, err) } } } diff --git a/hook_event_test.go b/hook_event_test.go new file mode 100644 index 0000000..ec755ea --- /dev/null +++ b/hook_event_test.go @@ -0,0 +1,471 @@ +package kgo + +import ( + "context" + "errors" + "io" + "net" + "os" + "testing" + + "github.com/stretchr/testify/require" + "github.com/twmb/franz-go/pkg/kgo" + "go.unistack.org/micro/v4/logger" +) + +func TestHookEvent_OnGroupManageError(t *testing.T) { + tests := []struct { + name string + inputErr error + fatalOnError bool + expectedErrorIsCalled bool + expectedErrorMsg string + expectedFatalIsCalled bool + expectedFatalMsg string + }{ + { + name: "error is nil", + inputErr: nil, + expectedErrorIsCalled: false, + expectedFatalIsCalled: false, + }, + { + name: "context canceled", + inputErr: context.Canceled, + expectedErrorIsCalled: false, + expectedFatalIsCalled: false, + }, + { + name: "context deadline exceeded", + inputErr: context.DeadlineExceeded, + expectedErrorIsCalled: false, + expectedFatalIsCalled: false, + }, + { + name: "retryable error: deadline exceeded (os package)", + inputErr: os.ErrDeadlineExceeded, + expectedErrorIsCalled: false, + expectedFatalIsCalled: false, + }, + { + name: "retryable error: EOF (io package)", + inputErr: io.EOF, + expectedErrorIsCalled: false, + expectedFatalIsCalled: false, + }, + { + name: "retryable error: closed network connection (net package)", + inputErr: net.ErrClosed, + expectedErrorIsCalled: false, + expectedFatalIsCalled: false, + }, + { + name: "some error (non-fatal)", + inputErr: errors.New("some error"), + fatalOnError: false, + expectedErrorIsCalled: true, + expectedErrorMsg: "kgo.OnGroupManageError", + expectedFatalIsCalled: false, + }, + { + name: "some error (fatal)", + inputErr: errors.New("some error"), + fatalOnError: true, + expectedErrorIsCalled: false, + expectedFatalIsCalled: true, + expectedFatalMsg: "kgo.OnGroupManageError", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + log := &mockLogger{} + he := &hookEvent{log: log, fatalOnError: tt.fatalOnError} + he.OnGroupManageError(tt.inputErr) + require.Equal(t, tt.expectedErrorIsCalled, log.errorIsCalled) + require.Equal(t, tt.expectedErrorMsg, log.errorMsg) + require.Equal(t, tt.expectedFatalIsCalled, log.fatalIsCalled) + require.Equal(t, tt.expectedFatalMsg, log.fatalMsg) + }) + } +} + +func TestHookEvent_OnBrokerConnect(t *testing.T) { + tests := []struct { + name string + inputErr error + fatalOnError bool + expectedErrorIsCalled bool + expectedErrorMsg string + expectedFatalIsCalled bool + expectedFatalMsg string + }{ + { + name: "error is nil", + inputErr: nil, + expectedErrorIsCalled: false, + expectedFatalIsCalled: false, + }, + { + name: "context canceled", + inputErr: context.Canceled, + expectedErrorIsCalled: false, + expectedFatalIsCalled: false, + }, + { + name: "context deadline exceeded", + inputErr: context.DeadlineExceeded, + expectedErrorIsCalled: false, + expectedFatalIsCalled: false, + }, + { + name: "retryable error: deadline exceeded (os package)", + inputErr: os.ErrDeadlineExceeded, + expectedErrorIsCalled: false, + expectedFatalIsCalled: false, + }, + { + name: "retryable error: EOF (io package)", + inputErr: io.EOF, + expectedErrorIsCalled: false, + expectedFatalIsCalled: false, + }, + { + name: "retryable error: closed network connection (net package)", + inputErr: net.ErrClosed, + expectedErrorIsCalled: false, + expectedFatalIsCalled: false, + }, + { + name: "some error (non-fatal)", + inputErr: errors.New("some error"), + fatalOnError: false, + expectedErrorIsCalled: true, + expectedErrorMsg: "kgo.OnBrokerConnect", + expectedFatalIsCalled: false, + }, + { + name: "some error (fatal)", + inputErr: errors.New("some error"), + fatalOnError: true, + expectedErrorIsCalled: false, + expectedFatalIsCalled: true, + expectedFatalMsg: "kgo.OnBrokerConnect", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + log := &mockLogger{} + he := &hookEvent{log: log, fatalOnError: tt.fatalOnError} + he.OnBrokerConnect(kgo.BrokerMetadata{}, 0, nil, tt.inputErr) + require.Equal(t, tt.expectedErrorIsCalled, log.errorIsCalled) + require.Equal(t, tt.expectedErrorMsg, log.errorMsg) + require.Equal(t, tt.expectedFatalIsCalled, log.fatalIsCalled) + require.Equal(t, tt.expectedFatalMsg, log.fatalMsg) + }) + } +} + +func TestHookEvent_OnBrokerWrite(t *testing.T) { + tests := []struct { + name string + inputErr error + fatalOnError bool + expectedErrorIsCalled bool + expectedErrorMsg string + expectedFatalIsCalled bool + expectedFatalMsg string + }{ + { + name: "error is nil", + inputErr: nil, + expectedErrorIsCalled: false, + expectedFatalIsCalled: false, + }, + { + name: "context canceled", + inputErr: context.Canceled, + expectedErrorIsCalled: false, + expectedFatalIsCalled: false, + }, + { + name: "context deadline exceeded", + inputErr: context.DeadlineExceeded, + expectedErrorIsCalled: false, + expectedFatalIsCalled: false, + }, + { + name: "retryable error: deadline exceeded (os package)", + inputErr: os.ErrDeadlineExceeded, + expectedErrorIsCalled: false, + expectedFatalIsCalled: false, + }, + { + name: "retryable error: EOF (io package)", + inputErr: io.EOF, + expectedErrorIsCalled: false, + expectedFatalIsCalled: false, + }, + { + name: "retryable error: closed network connection (net package)", + inputErr: net.ErrClosed, + expectedErrorIsCalled: false, + expectedFatalIsCalled: false, + }, + { + name: "some error (non-fatal)", + inputErr: errors.New("some error"), + fatalOnError: false, + expectedErrorIsCalled: true, + expectedErrorMsg: "kgo.OnBrokerWrite", + expectedFatalIsCalled: false, + }, + { + name: "some error (fatal)", + inputErr: errors.New("some error"), + fatalOnError: true, + expectedErrorIsCalled: false, + expectedFatalIsCalled: true, + expectedFatalMsg: "kgo.OnBrokerWrite", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + log := &mockLogger{} + he := &hookEvent{log: log, fatalOnError: tt.fatalOnError} + he.OnBrokerWrite(kgo.BrokerMetadata{}, 0, 0, 0, 0, tt.inputErr) + require.Equal(t, tt.expectedErrorIsCalled, log.errorIsCalled) + require.Equal(t, tt.expectedErrorMsg, log.errorMsg) + require.Equal(t, tt.expectedFatalIsCalled, log.fatalIsCalled) + require.Equal(t, tt.expectedFatalMsg, log.fatalMsg) + }) + } +} + +func TestHookEvent_OnBrokerRead(t *testing.T) { + tests := []struct { + name string + inputErr error + fatalOnError bool + expectedErrorIsCalled bool + expectedErrorMsg string + expectedFatalIsCalled bool + expectedFatalMsg string + }{ + { + name: "error is nil", + inputErr: nil, + expectedErrorIsCalled: false, + expectedFatalIsCalled: false, + }, + { + name: "context canceled", + inputErr: context.Canceled, + expectedErrorIsCalled: false, + expectedFatalIsCalled: false, + }, + { + name: "context deadline exceeded", + inputErr: context.DeadlineExceeded, + expectedErrorIsCalled: false, + expectedFatalIsCalled: false, + }, + { + name: "retryable error: deadline exceeded (os package)", + inputErr: os.ErrDeadlineExceeded, + expectedErrorIsCalled: false, + expectedFatalIsCalled: false, + }, + { + name: "retryable error: EOF (io package)", + inputErr: io.EOF, + expectedErrorIsCalled: false, + expectedFatalIsCalled: false, + }, + { + name: "retryable error: closed network connection (net package)", + inputErr: net.ErrClosed, + expectedErrorIsCalled: false, + expectedFatalIsCalled: false, + }, + { + name: "some error (non-fatal)", + inputErr: errors.New("some error"), + fatalOnError: false, + expectedErrorIsCalled: true, + expectedErrorMsg: "kgo.OnBrokerRead", + expectedFatalIsCalled: false, + }, + { + name: "some error (fatal)", + inputErr: errors.New("some error"), + fatalOnError: true, + expectedErrorIsCalled: false, + expectedFatalIsCalled: true, + expectedFatalMsg: "kgo.OnBrokerRead", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + log := &mockLogger{} + he := &hookEvent{log: log, fatalOnError: tt.fatalOnError} + he.OnBrokerRead(kgo.BrokerMetadata{}, 0, 0, 0, 0, tt.inputErr) + require.Equal(t, tt.expectedErrorIsCalled, log.errorIsCalled) + require.Equal(t, tt.expectedErrorMsg, log.errorMsg) + require.Equal(t, tt.expectedFatalIsCalled, log.fatalIsCalled) + require.Equal(t, tt.expectedFatalMsg, log.fatalMsg) + }) + } +} + +func TestHookEvent_OnProduceRecordUnbuffered(t *testing.T) { + tests := []struct { + name string + inputErr error + fatalOnError bool + expectedErrorIsCalled bool + expectedErrorMsg string + expectedFatalIsCalled bool + expectedFatalMsg string + }{ + { + name: "error is nil", + inputErr: nil, + expectedErrorIsCalled: false, + expectedFatalIsCalled: false, + }, + { + name: "context canceled", + inputErr: context.Canceled, + expectedErrorIsCalled: false, + expectedFatalIsCalled: false, + }, + { + name: "context deadline exceeded", + inputErr: context.DeadlineExceeded, + expectedErrorIsCalled: false, + expectedFatalIsCalled: false, + }, + { + name: "retryable error: deadline exceeded (os package)", + inputErr: os.ErrDeadlineExceeded, + expectedErrorIsCalled: false, + expectedFatalIsCalled: false, + }, + { + name: "retryable error: EOF (io package)", + inputErr: io.EOF, + expectedErrorIsCalled: false, + expectedFatalIsCalled: false, + }, + { + name: "retryable error: closed network connection (net package)", + inputErr: net.ErrClosed, + expectedErrorIsCalled: false, + expectedFatalIsCalled: false, + }, + { + name: "some error (non-fatal)", + inputErr: errors.New("some error"), + fatalOnError: false, + expectedErrorIsCalled: true, + expectedErrorMsg: "kgo.OnProduceRecordUnbuffered", + expectedFatalIsCalled: false, + }, + { + name: "some error (fatal)", + inputErr: errors.New("some error"), + fatalOnError: true, + expectedErrorIsCalled: false, + expectedFatalIsCalled: true, + expectedFatalMsg: "kgo.OnProduceRecordUnbuffered", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + log := &mockLogger{} + he := &hookEvent{log: log, fatalOnError: tt.fatalOnError} + he.OnProduceRecordUnbuffered(&kgo.Record{}, tt.inputErr) + require.Equal(t, tt.expectedErrorIsCalled, log.errorIsCalled) + require.Equal(t, tt.expectedErrorMsg, log.errorMsg) + require.Equal(t, tt.expectedFatalIsCalled, log.fatalIsCalled) + require.Equal(t, tt.expectedFatalMsg, log.fatalMsg) + }) + } +} + +// Mocks + +type mockLogger struct { + errorIsCalled bool + errorMsg string + + fatalIsCalled bool + fatalMsg string +} + +func (m *mockLogger) Init(...logger.Option) error { + panic("implement me") +} + +func (m *mockLogger) Clone(...logger.Option) logger.Logger { + panic("implement me") +} + +func (m *mockLogger) V(logger.Level) bool { + panic("implement me") +} + +func (m *mockLogger) Level(logger.Level) { + panic("implement me") +} + +func (m *mockLogger) Options() logger.Options { + panic("implement me") +} + +func (m *mockLogger) Fields(...interface{}) logger.Logger { + panic("implement me") +} + +func (m *mockLogger) Info(context.Context, string, ...interface{}) { + panic("implement me") +} + +func (m *mockLogger) Trace(context.Context, string, ...interface{}) { + panic("implement me") +} + +func (m *mockLogger) Debug(context.Context, string, ...interface{}) { + panic("implement me") +} + +func (m *mockLogger) Warn(context.Context, string, ...interface{}) { + panic("implement me") +} + +func (m *mockLogger) Error(ctx context.Context, msg string, args ...interface{}) { + m.errorIsCalled = true + m.errorMsg = msg +} + +func (m *mockLogger) Fatal(ctx context.Context, msg string, args ...interface{}) { + m.fatalIsCalled = true + m.fatalMsg = msg +} + +func (m *mockLogger) Log(context.Context, logger.Level, string, ...interface{}) { + panic("implement me") +} + +func (m *mockLogger) Name() string { + panic("implement me") +} + +func (m *mockLogger) String() string { + panic("implement me") +}