{"record":{"id":"7213e3de32bce95a","repo":"crowdsecurity/crowdsec","slug":"while-closing-s-reader-on-topic-s-w","errorCode":null,"errorMessage":"while closing %s reader on topic '%s': %w","messagePattern":"while closing (.+?) reader on topic '(.+?)': %w","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"warning","filePath":"pkg/acquisition/modules/kafka/run.go","lineNumber":74,"sourceCode":"\t\tevt := pipeline.MakeEvent(s.Config.UseTimeMachine, pipeline.LOG, true)\n\t\tevt.Line = l\n\n\t\tout <- evt\n\t}\n}\n\nfunc (s *Source) RunReader(ctx context.Context, out chan pipeline.Event, t *tomb.Tomb) error {\n\ts.logger.Debugf(\"starting %s datasource reader goroutine with configuration %+v\", s.GetName(), s.Config)\n\tt.Go(func() error {\n\t\treturn s.ReadMessage(ctx, out)\n\t})\n\n\t<-t.Dying()\n\n\ts.logger.Infof(\"%s datasource topic %s stopping\", s.GetName(), s.Config.Topic)\n\n\tif err := s.Reader.Close(); err != nil {\n\t\treturn fmt.Errorf(\"while closing %s reader on topic '%s': %w\", s.GetName(), s.Config.Topic, err)\n\t}\n\n\treturn nil\n}\n\nfunc (s *Source) StreamingAcquisition(ctx context.Context, out chan pipeline.Event, t *tomb.Tomb) error {\n\ts.logger.Infof(\"start reader on brokers '%+v' with topic '%s'\", s.Config.Brokers, s.Config.Topic)\n\n\tt.Go(func() error {\n\t\tdefer trace.ReportPanic()\n\t\treturn s.RunReader(ctx, out, t)\n\t})\n\n\treturn nil\n}\n","sourceCodeStart":56,"sourceCodeEnd":90,"githubUrl":"https://github.com/crowdsecurity/crowdsec/blob/909b5157986a2b2c2163300fdaef5ed01289f7d2/pkg/acquisition/modules/kafka/run.go#L56-L90","documentation":"Returned by RunReader when s.Reader.Close() fails during graceful shutdown of the kafka datasource (the tomb signals dying). Close flushes and releases connections; a failure here can mean leaked connections or uncommitted state, but the datasource is already shutting down.","triggerScenarios":"CrowdSec stops/reloads the kafka datasource; the tomb fires and s.Reader.Close() returns an error — commonly because the reader is mid-request, the broker is unreachable, or Close is called on an already-closed reader.","commonSituations":"Broker down at shutdown time; concurrent ReadMessage in flight while Close is called; double Close after a restart loop; slow network causing Close's internal deadline to expire.","solutions":["Ensure no concurrent ReadMessage call is active when Close is invoked (cancel/drain the read loop first)","Check broker availability during shutdown; retry the reload once the broker is back","Inspect the wrapped %w error: 'reader was closed' indicates double Close — fix the lifecycle to close once","As a last resort this is usually harmless at process exit; treat as noise only after confirming connections are cleaned up"],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"if err := reader.Close(); err != nil {\n    if strings.Contains(err.Error(), \"closed\") {\n        return nil // double-close, ignore\n    }\n    logger.Warnf(\"kafka reader close during shutdown: %v\", err)\n}","preventionTips":["Close the reader from exactly one place in the lifecycle","Cancel/finish in-flight ReadMessage before Close","Log as warning, not error, during shutdown paths"],"tags":["kafka","shutdown","kafka-go","crowdsec"],"backgroundTag":"invalid-state-transition","analyzedSha":"909b5157986a2b2c2163300fdaef5ed01289f7d2","analyzedAt":"2026-09-06T12:27:26.012Z","contentChangedAt":"2026-09-06T12:27:26.012Z","schemaVersion":2},"datasetVersion":"2026-09-14T00:17:10.932Z"}