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

Merge pull request 'correction create opts' (#159) from devstigneev/micro-broker-kgo:v3_fix_opts into v3
Some checks failed
test / test (push) Failing after 17m33s
Details
coverage / build (push) Failing after 17m44s
Details

Browse Source 
Reviewed-on: #159
This commit was merged in pull request #159.
This commit is contained in:
Василий Толстов
2026-01-30 09:48:26 +03:00
parent c8eeb34efe 79d80a6fc8
commit 8a069e9faf
1 changed files with 16 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
kgo.go
View File

@@ -477,11 +477,12 @@ func (k *Broker) Subscribe(ctx context.Context, topic string, handler broker.Han
connected: k.connected, connected: k.connected,
} }
kopts := append(k.kopts, kopts := append(
[]kgo.Opt{
kgo.ConsumerGroup(options.Group), kgo.ConsumerGroup(options.Group),
kgo.ConsumeTopics(topic), kgo.ConsumeTopics(topic),
kgo.ConsumeResetOffset(kgo.NewOffset().AtStart()), kgo.ConsumeResetOffset(kgo.NewOffset().AtStart()),
kgo.FetchMaxWait(1*time.Second), kgo.FetchMaxWait(1 * time.Second),
kgo.AutoCommitInterval(commitInterval), kgo.AutoCommitInterval(commitInterval),
kgo.OnPartitionsAssigned(sub.assigned), kgo.OnPartitionsAssigned(sub.assigned),
kgo.OnPartitionsRevoked(sub.revoked), kgo.OnPartitionsRevoked(sub.revoked),
@@ -490,6 +491,8 @@ func (k *Broker) Subscribe(ctx context.Context, topic string, handler broker.Han
kgo.AutoCommitCallback(sub.autocommit), kgo.AutoCommitCallback(sub.autocommit),
kgo.AutoCommitMarks(), kgo.AutoCommitMarks(),
kgo.WithHooks(sub), kgo.WithHooks(sub),
},
k.kopts...,
) )
if options.Context != nil { if options.Context != nil {