johanels/akvorado
1
0
Fork 0
mirror of https://github.com/akvorado/akvorado.git synced 2025-12-11 22:14:02 +01:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

outlet/kafka: delay shutdown when stopping worker after closing client
Some checks failed
CI / 🤖 Check dependabot status (push) Has been cancelled
Details
CI / 🐧 Test on Linux (${{ github.ref_type == 'tag' }}, misc) (push) Has been cancelled
Details
CI / 🐧 Test on Linux (coverage) (push) Has been cancelled
Details
CI / 🐧 Test on Linux (regular) (push) Has been cancelled
Details
CI / ❄️ Build on Nix (push) Has been cancelled
Details
CI / 🍏 Build and test on macOS (push) Has been cancelled
Details
CI / 🧪 End-to-end testing (push) Has been cancelled
Details
CI / 🔍 Upload code coverage (push) Has been cancelled
Details
CI / 🔬 Test only Go (push) Has been cancelled
Details
CI / 🔬 Test only JS (${{ needs.dependabot.outputs.package-ecosystem }}, 20) (push) Has been cancelled
Details
CI / 🔬 Test only JS (${{ needs.dependabot.outputs.package-ecosystem }}, 22) (push) Has been cancelled
Details
CI / 🔬 Test only JS (${{ needs.dependabot.outputs.package-ecosystem }}, 24) (push) Has been cancelled
Details
CI / ⚖️ Check licenses (push) Has been cancelled
Details
CI / 🐋 Build Docker images (push) Has been cancelled
Details
CI / 🐋 Tag Docker images (push) Has been cancelled
Details
CI / 🚀 Publish release (push) Has been cancelled
Details

Browse Source 
Flushing can take some time and we have an heartbeat to respect to
commit offsets.
This commit is contained in:
Vincent Bernat
2025-11-13 22:53:34 +01:00
parent 0028f2b524
commit 3291abe680
1 changed files with 2 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
outlet/kafka/worker.go
View File

@@ -66,7 +66,6 @@ func (c *realComponent) startOneWorker() error {
Logger()
defer func() {
logger.Info().Msg("stopping worker")
shutdown()
// Allow a small grace time to commit uncommited work.
ctx, cancelCommit := context.WithTimeout(context.Background(), 10*time.Second)
@@ -74,8 +73,9 @@ func (c *realComponent) startOneWorker() error {
if err := client.CommitMarkedOffsets(ctx); err != nil {
logger.Err(err).Msg("cannot commit marked partition offsets")
}
client.CloseAllowingRebalance()
shutdown()
}()
for {