Compare commits

..

19 Commits
v4.1.6 ... v4

Author SHA1 Message Date
vtolstov
c4f8de4aca Apply Code Coverage Badge 2025-07-15 13:02:27 +00:00
63d82be92a add additional metadata to message, closes #194
Some checks failed
sync / sync (push) Failing after 12m3s
coverage / build (push) Failing after 13m41s
test / test (push) Successful in 4m0s
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2025-07-15 15:59:05 +03:00
cced3b84c1 remove noisy logs from event hooks and add tests (#193)
Some checks failed
coverage / build (push) Failing after 1m50s
test / test (push) Failing after 19m17s
sync / sync (push) Successful in 7s
2025-07-10 12:51:26 +03:00
vtolstov
82c1bda50f Apply Code Coverage Badge 2025-07-02 13:21:33 +00:00
4b0b70e7d0 fixup message options and content-type header
Some checks failed
coverage / build (push) Successful in 4m34s
test / test (push) Failing after 17m28s
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2025-07-02 16:20:02 +03:00
vtolstov
da88718e03 Apply Code Coverage Badge 2025-06-30 13:11:01 +00:00
aed304b7e3 rework message options
Some checks failed
sync / sync (push) Successful in 1m2s
coverage / build (push) Successful in 3m31s
test / test (push) Failing after 18m28s
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2025-06-30 16:08:59 +03:00
b796c43257 [v4] hide access to internal mutex (#185)
Some checks failed
coverage / build (push) Failing after 2m4s
test / test (push) Successful in 3m53s
sync / sync (push) Successful in 8s
* changed embedded mutex to private field

* update ci

* fix tests
2025-06-10 16:28:17 +03:00
b0691edf82 disable message pool by default
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2025-06-10 16:28:16 +03:00
d43b59f2f3 disable message pool by default
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2025-06-10 16:28:03 +03:00
f976516d6b new kgo version
update deps

fixup race conditions

add kfake usage

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2025-06-10 16:27:40 +03:00
8f5c03ad60 [v4] hide access to internal mutex (#185)
* changed embedded mutex to private field

* update ci

* fix tests
2025-06-10 16:26:52 +03:00
vtolstov
9e9c1434ff Apply Code Coverage Badge 2025-06-09 14:30:14 +00:00
fe81a14413 update for latest micro
Some checks failed
coverage / build (push) Successful in 3m22s
test / test (push) Failing after 19m17s
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2025-06-09 17:27:45 +03:00
eb56a48dda update kgo
Some checks failed
coverage / build (push) Has been cancelled
test / test (push) Has been cancelled
sync / sync (push) Failing after 28s
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2025-05-25 01:25:56 +03:00
71e8a13c70 new kgo version
Some checks failed
sync / sync (push) Has been cancelled
coverage / build (push) Failing after 1m31s
test / test (push) Failing after 18m29s
update deps

fixup race conditions

add kfake usage

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2025-05-25 01:24:50 +03:00
b538ef82b5 [v4] hide access to internal mutex (#185)
* changed embedded mutex to private field

* update ci

* fix tests
2025-05-25 01:23:46 +03:00
44e2d5f9a4 update ci (#188) 2025-05-25 01:23:11 +03:00
vtolstov
9426208e1c Apply Code Coverage Badge 2025-05-21 08:14:38 +00:00
14 changed files with 716 additions and 139 deletions

View File

@@ -8,12 +8,11 @@ on:
- '.gitea/**'
pull_request:
branches: [ main, v3, v4 ]
# Allows you to run this workflow manually from the Actions tab
workflow_dispatch:
jobs:
build:
if: github.server_url != 'https://github.com'
runs-on: ubuntu-latest
steps:
- name: checkout code
@@ -42,8 +41,8 @@ jobs:
name: autocommit
with:
commit_message: Apply Code Coverage Badge
skip_fetch: true
skip_checkout: true
skip_fetch: false
skip_checkout: false
file_pattern: ./README.md
- name: push

View File

@@ -24,6 +24,6 @@ jobs:
- name: setup deps
run: go get -v ./...
- name: run lint
uses: https://github.com/golangci/golangci-lint-action@v6
uses: golangci/golangci-lint-action@v6
with:
version: 'latest'

View File

@@ -8,7 +8,7 @@ on:
jobs:
sync:
if: github.server_url == 'zhttps://github.com'
if: github.server_url != 'https://github.com'
runs-on: ubuntu-latest
steps:
- name: init
@@ -18,34 +18,76 @@ jobs:
echo "machine git.unistack.org login vtolstov password ${{ secrets.TOKEN_GITEA }}" >> /root/.netrc
echo "machine github.com login vtolstov password ${{ secrets.TOKEN_GITHUB }}" >> /root/.netrc
- name: check master
id: check_master
run: |
src_hash=$(git ls-remote https://github.com/${GITHUB_REPOSITORY} refs/heads/master | cut -f1)
dst_hash=$(git ls-remote ${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY} refs/heads/master | cut -f1)
echo "src_hash=$src_hash"
echo "dst_hash=$dst_hash"
if [ "$src_hash" != "$dst_hash" ]; then
echo "sync_needed=true" >> $GITHUB_OUTPUT
else
echo "sync_needed=false" >> $GITHUB_OUTPUT
fi
- name: sync master
if: steps.check_master.outputs.sync_needed == 'true'
run: |
git clone --filter=blob:none --filter=tree:0 --branch master --single-branch ${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY} repo
cd repo
git remote add --no-tags --fetch --track master upstream https://github.com/${GITHUB_REPOSITORY}
git merge upstream/master
git pull --rebase upstream master
git push upstream master --progress
git push origin master --progress
cd ../
rm -rf repo
- name: check v3
id: check_v3
run: |
src_hash=$(git ls-remote https://github.com/${GITHUB_REPOSITORY} refs/heads/v3 | cut -f1)
dst_hash=$(git ls-remote ${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY} refs/heads/v3 | cut -f1)
echo "src_hash=$src_hash"
echo "dst_hash=$dst_hash"
if [ "$src_hash" != "$dst_hash" ]; then
echo "sync_needed=true" >> $GITHUB_OUTPUT
else
echo "sync_needed=false" >> $GITHUB_OUTPUT
fi
- name: sync v3
if: steps.check_v3.outputs.sync_needed == 'true'
run: |
git clone --filter=blob:none --filter=tree:0 --branch v3 --single-branch ${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY} repo
cd repo
git remote add --no-tags --fetch --track v3 upstream https://github.com/${GITHUB_REPOSITORY}
git merge upstream/v3
git pull --rebase upstream v3
git push upstream v3 --progress
git push origin v3 --progress
cd ../
rm -rf repo
- name: check v4
id: check_v4
run: |
src_hash=$(git ls-remote https://github.com/${GITHUB_REPOSITORY} refs/heads/v4 | cut -f1)
dst_hash=$(git ls-remote ${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY} refs/heads/v4 | cut -f1)
echo "src_hash=$src_hash"
echo "dst_hash=$dst_hash"
if [ "$src_hash" != "$dst_hash" ]; then
echo "sync_needed=true" >> $GITHUB_OUTPUT
else
echo "sync_needed=false" >> $GITHUB_OUTPUT
fi
- name: sync v4
if: steps.check_v4.outputs.sync_needed == 'true'
run: |
git clone --filter=blob:none --filter=tree:0 --branch v4 --single-branch ${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY} repo
cd repo
git remote add --no-tags --fetch --track v4 upstream https://github.com/${GITHUB_REPOSITORY}
git merge upstream/v4
git pull --rebase upstream v4
git push upstream v4 --progress
git push origin v4 --progress
cd ../

View File

@@ -32,19 +32,19 @@ jobs:
go-version: 'stable'
- name: setup go work
env:
GOWORK: /workspace/${{ github.repository_owner }}/go.work
GOWORK: ${{ github.workspace }}/go.work
run: |
go work init
go work use .
go work use micro-tests
- name: setup deps
env:
GOWORK: /workspace/${{ github.repository_owner }}/go.work
GOWORK: ${{ github.workspace }}/go.work
run: go get -v ./...
- name: run tests
env:
INTEGRATION_TESTS: yes
GOWORK: /workspace/${{ github.repository_owner }}/go.work
GOWORK: ${{ github.workspace }}/go.work
run: |
cd micro-tests
go test -mod readonly -v ./... || true

View File

@@ -1,2 +1,2 @@
# micro-broker-kgo
![Coverage](https://img.shields.io/badge/Coverage-65.8%25-yellow)
![Coverage](https://img.shields.io/badge/Coverage-66.9%25-yellow)

10
errors.go Normal file
View File

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

14
go.mod
View File

@@ -3,21 +3,25 @@ module go.unistack.org/micro-broker-kgo/v4
go 1.23.8
require (
github.com/twmb/franz-go v1.19.1
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
github.com/twmb/franz-go/pkg/kmsg v1.11.2
go.opentelemetry.io/otel v1.35.0
go.unistack.org/micro/v4 v4.1.14
go.opentelemetry.io/otel v1.36.0
go.unistack.org/micro/v4 v4.1.17
)
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/spf13/cast v1.8.0 // 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.38.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
)

23
go.sum
View File

@@ -20,27 +20,30 @@ 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.8.0 h1:gEN9K4b8Xws4EX0+a0reLmhq8moKn7ntRlQYgjPeCDk=
github.com/spf13/cast v1.8.0/go.mod h1:ancEpBxwJDODSW/UG4rDrAqiKolqNNh2DX3mk86cAdo=
github.com/spf13/cast v1.9.2 h1:SsGfm7M8QOFtEzumm7UZrZdLLquNdzFYfIbEXntcFbE=
github.com/spf13/cast v1.9.2/go.mod h1:jNfB8QC9IA6ZuY2ZjDp0KtFO2LZZlg4S/7bzP6qqeHo=
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.19.1 h1:cOhDFUkGvUFHSQ7UYW6bO77BJa2fYEk5mA2AX+1NIdE=
github.com/twmb/franz-go v1.19.1/go.mod h1:4kFJ5tmbbl7asgwAGVuyG1ZMx0NNpYk7EqflvWfPCpM=
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/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.opentelemetry.io/otel v1.36.0 h1:UumtzIklRBY6cI/lllNZlALOF5nNIzJVb16APdvgTXg=
go.opentelemetry.io/otel v1.36.0/go.mod h1:/TcFMXYjyRNh8khOAO9ybYkqaDBb/70aVwkNML4pP8E=
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.14 h1:6EotPq9kz/gaFb5YulHdKuuUwmj/7Hk44DpOlzh/A6k=
go.unistack.org/micro/v4 v4.1.14/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=
go.unistack.org/micro/v4 v4.1.17 h1:26QDtRSYVpozYuassyvLP4sEQRo3dxgD3sVILRXmIPo=
go.unistack.org/micro/v4 v4.1.17/go.mod h1:xleO2M5Yxh4s6I+RUcLrEpUjobefh+71ctrdIfn7TUs=
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=

View File

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

471
hook_event_test.go Normal file
View File

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

122
kgo.go
View File

@@ -74,7 +74,7 @@ type Broker struct {
opts broker.Options
sync.RWMutex
mu sync.RWMutex
init bool
}
@@ -108,7 +108,7 @@ type kgoMessage struct {
ctx context.Context
body []byte
hdr metadata.Metadata
opts broker.PublishOptions
opts broker.MessageOptions
ack bool
}
@@ -141,22 +141,22 @@ func (b *Broker) newCodec(ct string) (codec.Codec, error) {
if idx := strings.IndexRune(ct, ';'); idx >= 0 {
ct = ct[:idx]
}
b.RLock()
b.mu.RLock()
c, ok := b.opts.Codecs[ct]
b.RUnlock()
b.mu.RUnlock()
if ok {
return c, nil
}
return nil, codec.ErrUnknownContentType
}
func (b *Broker) NewMessage(ctx context.Context, hdr metadata.Metadata, body interface{}, opts ...broker.PublishOption) (broker.Message, error) {
options := broker.NewPublishOptions(opts...)
func (b *Broker) NewMessage(ctx context.Context, hdr metadata.Metadata, body interface{}, opts ...broker.MessageOption) (broker.Message, error) {
options := broker.NewMessageOptions(opts...)
if options.ContentType == "" {
options.ContentType = b.opts.ContentType
}
m := &kgoMessage{ctx: ctx, hdr: hdr, opts: options}
m := &kgoMessage{ctx: ctx, hdr: hdr.Copy(), opts: options}
c, err := b.newCodec(m.opts.ContentType)
if err == nil {
m.body, err = c.Marshal(body)
@@ -165,6 +165,8 @@ func (b *Broker) NewMessage(ctx context.Context, hdr metadata.Metadata, body int
return nil, err
}
m.hdr.Set(metadata.HeaderContentType, m.opts.ContentType)
return m, nil
}
@@ -238,10 +240,10 @@ 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()
return nil
}
@@ -259,8 +261,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()
@@ -284,8 +286,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
@@ -344,40 +346,44 @@ func (b *Broker) fnPublish(ctx context.Context, topic string, messages ...broker
}
func (b *Broker) publish(ctx context.Context, topic string, messages ...broker.Message) error {
records := make([]*kgo.Record, 0, len(messages))
var errs []string
var key []byte
var promise func(*kgo.Record, error)
var records []*kgo.Record
for _, msg := range messages {
if mctx := msg.Context(); mctx != nil {
if k, ok := mctx.Value(publishKey{}).([]byte); ok && k != nil {
key = k
}
if p, ok := mctx.Value(publishPromiseKey{}).(func(*kgo.Record, error)); ok && p != nil {
promise = p
}
}
rec := &kgo.Record{
Context: ctx,
Key: key,
Context: msg.Context(),
Topic: topic,
Value: msg.Body(),
}
b.opts.Meter.Counter(semconv.PublishMessageInflight, "endpoint", rec.Topic, "topic", rec.Topic).Inc()
var promise func(*kgo.Record, error)
if rec.Context != nil {
if k, ok := rec.Context.Value(messageKey{}).([]byte); ok && k != nil {
rec.Key = k
}
if p, ok := rec.Context.Value(messagePromiseKey{}).(func(*kgo.Record, error)); ok && p != nil {
promise = p
}
}
kmsg, ok := msg.(*kgoMessage)
if !ok {
continue
}
if kmsg.opts.Context != nil {
if k, ok := kmsg.opts.Context.Value(messageKey{}).([]byte); ok && k != nil {
rec.Key = k
}
if p, ok := kmsg.opts.Context.Value(messagePromiseKey{}).(func(*kgo.Record, error)); ok && p != nil {
promise = p
}
}
setHeaders(rec, msg.Header())
records = append(records, rec)
}
ts := time.Now()
if promise != nil {
for _, rec := range records {
if promise != nil {
ts := time.Now()
b.opts.Meter.Counter(semconv.PublishMessageInflight, "endpoint", rec.Topic, "topic", rec.Topic).Inc()
b.c.Produce(ctx, rec, func(r *kgo.Record, err error) {
te := time.Since(ts)
b.opts.Meter.Counter(semconv.PublishMessageInflight, "endpoint", rec.Topic, "topic", rec.Topic).Dec()
@@ -390,27 +396,33 @@ func (b *Broker) publish(ctx context.Context, topic string, messages ...broker.M
}
promise(r, err)
})
}
return nil
}
results := b.c.ProduceSync(ctx, records...)
te := time.Since(ts)
for _, result := range results {
b.opts.Meter.Summary(semconv.PublishMessageLatencyMicroseconds, "endpoint", result.Record.Topic, "topic", result.Record.Topic).Update(te.Seconds())
b.opts.Meter.Histogram(semconv.PublishMessageDurationSeconds, "endpoint", result.Record.Topic, "topic", result.Record.Topic).Update(te.Seconds())
b.opts.Meter.Counter(semconv.PublishMessageInflight, "endpoint", result.Record.Topic, "topic", result.Record.Topic).Dec()
if result.Err != nil {
b.opts.Meter.Counter(semconv.PublishMessageTotal, "endpoint", result.Record.Topic, "topic", result.Record.Topic, "status", "failure").Inc()
errs = append(errs, result.Err.Error())
continue
} else {
b.opts.Meter.Counter(semconv.PublishMessageTotal, "endpoint", result.Record.Topic, "topic", result.Record.Topic, "status", "success").Inc()
records = append(records, rec)
}
}
if len(errs) > 0 {
return fmt.Errorf("publish error: %s", strings.Join(errs, "\n"))
if len(records) > 0 {
var errs []string
ts := time.Now()
b.opts.Meter.Counter(semconv.PublishMessageInflight, "endpoint", topic, "topic", topic).Set(uint64(len(records)))
results := b.c.ProduceSync(ctx, records...)
te := time.Since(ts)
for _, result := range results {
b.opts.Meter.Summary(semconv.PublishMessageLatencyMicroseconds, "endpoint", result.Record.Topic, "topic", result.Record.Topic).Update(te.Seconds())
b.opts.Meter.Histogram(semconv.PublishMessageDurationSeconds, "endpoint", result.Record.Topic, "topic", result.Record.Topic).Update(te.Seconds())
b.opts.Meter.Counter(semconv.PublishMessageInflight, "endpoint", result.Record.Topic, "topic", result.Record.Topic).Dec()
if result.Err != nil {
b.opts.Meter.Counter(semconv.PublishMessageTotal, "endpoint", result.Record.Topic, "topic", result.Record.Topic, "status", "failure").Inc()
errs = append(errs, result.Err.Error())
} else {
b.opts.Meter.Counter(semconv.PublishMessageTotal, "endpoint", result.Record.Topic, "topic", result.Record.Topic, "status", "success").Inc()
}
}
if len(errs) > 0 {
return fmt.Errorf("publish error: %s", strings.Join(errs, "\n"))
}
}
return nil
@@ -538,9 +550,9 @@ func (b *Broker) fnSubscribe(ctx context.Context, topic string, handler interfac
go sub.poll(ctx)
b.Lock()
b.mu.Lock()
b.subs = append(b.subs, sub)
b.Unlock()
b.mu.Unlock()
return sub, nil
}

View File

@@ -81,7 +81,7 @@ func TestFail(t *testing.T) {
go func() {
for _, msg := range msgs {
// t.Logf("broker publish")
if err := b.Publish(ctx, "test", msg); err != nil {
if err := b.Publish(ctx, "test.fail", msg); err != nil {
t.Fatal(err)
}
}
@@ -96,7 +96,7 @@ func TestFail(t *testing.T) {
return msg.Ack()
}
sub, err := b.Subscribe(ctx, "test", fn,
sub, err := b.Subscribe(ctx, "test.fail", fn,
broker.SubscribeAutoAck(true),
broker.SubscribeGroup(group),
broker.SubscribeBodyOnly(true))
@@ -184,7 +184,7 @@ func TestPubSub(t *testing.T) {
msgs = append(msgs, m)
}
if err := b.Publish(ctx, "test", msgs...); err != nil {
if err := b.Publish(ctx, "test.pubsub", msgs...); err != nil {
t.Fatal(err)
}
// t.Skip()
@@ -197,7 +197,7 @@ func TestPubSub(t *testing.T) {
return msg.Ack()
}
sub, err := b.Subscribe(ctx, "test", fn,
sub, err := b.Subscribe(ctx, "test.pubsub", fn,
broker.SubscribeAutoAck(true),
broker.SubscribeGroup(group),
broker.SubscribeBodyOnly(true))

View File

@@ -27,11 +27,11 @@ func SubscribeContext(ctx context.Context) broker.SubscribeOption {
return broker.SetSubscribeOption(subscribeContextKey{}, ctx)
}
type publishKey struct{}
type messageKey struct{}
// PublishKey set the kafka message key (broker option)
func PublishKey(key []byte) broker.PublishOption {
return broker.SetPublishOption(publishKey{}, key)
// MessageKey set the kafka message key (broker option)
func MessageKey(key []byte) broker.MessageOption {
return broker.SetMessageOption(messageKey{}, key)
}
type optionsKey struct{}
@@ -103,11 +103,11 @@ func SubscribeFatalOnError(b bool) broker.SubscribeOption {
return broker.SetSubscribeOption(fatalOnErrorKey{}, b)
}
type publishPromiseKey struct{}
type messagePromiseKey struct{}
// PublishPromise set the kafka promise func for Produce
func PublishPromise(fn func(*kgo.Record, error)) broker.PublishOption {
return broker.SetPublishOption(publishPromiseKey{}, fn)
// MessagePromise set the kafka promise func for Produce
func MessagePromise(fn func(*kgo.Record, error)) broker.MessageOption {
return broker.SetMessageOption(messagePromiseKey{}, fn)
}
type subscribeMessagePoolKey struct{}

View File

@@ -39,18 +39,17 @@ type consumer struct {
}
type Subscriber struct {
consumers map[tp]*consumer
c *kgo.Client
htracer *hookTracer
topic string
messagePool bool
handler interface{}
done chan struct{}
kopts broker.Options
opts broker.SubscribeOptions
connected *atomic.Uint32
sync.RWMutex
consumers map[tp]*consumer
c *kgo.Client
htracer *hookTracer
topic string
messagePool bool
handler interface{}
done chan struct{}
kopts broker.Options
opts broker.SubscribeOptions
connected *atomic.Uint32
mu sync.RWMutex
closed bool
fatalOnError bool
}
@@ -118,11 +117,11 @@ func (s *Subscriber) poll(ctx context.Context) {
continue
}
s.Lock()
s.mu.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()
s.mu.Unlock()
}
}
@@ -147,9 +146,9 @@ func (s *Subscriber) poll(ctx context.Context) {
fetches.EachPartition(func(p kgo.FetchTopicPartition) {
tps := tp{p.Topic, p.Partition}
s.Lock()
s.mu.Lock()
c := s.consumers[tps]
s.Unlock()
s.mu.Unlock()
if c != nil {
c.recs <- p
}
@@ -166,15 +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()
s.mu.Lock()
pc, ok := s.consumers[tps]
s.Unlock()
s.mu.Unlock()
if !ok {
continue
}
s.Lock()
s.mu.Lock()
delete(s.consumers, tps)
s.Unlock()
s.mu.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))
@@ -230,9 +229,9 @@ func (s *Subscriber) assigned(_ context.Context, c *kgo.Client, assigned map[str
opts: s.opts,
connected: s.connected,
}
s.Lock()
s.mu.Lock()
s.consumers[tp{topic, partition}] = pc
s.Unlock()
s.mu.Unlock()
go pc.consume()
}
}
@@ -272,7 +271,11 @@ func (pc *consumer) consume() {
for _, hdr := range record.Headers {
pm.hdr.Set(hdr.Key, string(hdr.Value))
}
pm.hdr.Set("Micro-Offset", strconv.FormatInt(record.Offset, 10))
pm.hdr.Set("Micro-Partition", strconv.FormatInt(int64(record.Partition), 10))
pm.hdr.Set("Micro-Topic", record.Topic)
pm.hdr.Set("Micro-Key", string(record.Key))
pm.hdr.Set("Micro-Timestamp", strconv.FormatInt(record.Timestamp.Unix(), 10))
switch h := pc.handler.(type) {
case func(broker.Message) error:
err = h(pm)