{"record":{"id":"ed59d4eb988231ea","repo":"crowdsecurity/crowdsec","slug":"while-setting-offset-for-reader-on-topic-s-w","errorCode":null,"errorMessage":"while setting offset for reader on topic '%s': %w","messagePattern":"while setting offset for reader on topic '(.+?)': %w","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"pkg/acquisition/modules/kafka/run.go","lineNumber":23,"sourceCode":"\t\"errors\"\n\t\"fmt\"\n\t\"io\"\n\n\t\"github.com/prometheus/client_golang/prometheus\"\n\t\"github.com/segmentio/kafka-go\"\n\t\"gopkg.in/tomb.v2\"\n\n\t\"github.com/crowdsecurity/go-cs-lib/trace\"\n\n\t\"github.com/crowdsecurity/crowdsec/pkg/metrics\"\n\t\"github.com/crowdsecurity/crowdsec/pkg/pipeline\"\n)\n\nfunc (s *Source) ReadMessage(ctx context.Context, out chan pipeline.Event) error {\n\tif s.Config.GroupID == \"\" {\n\t\terr := s.Reader.SetOffset(kafka.LastOffset)\n\t\tif err != nil {\n\t\t\treturn fmt.Errorf(\"while setting offset for reader on topic '%s': %w\", s.Config.Topic, err)\n\t\t}\n\t}\n\n\tfor {\n\t\ts.logger.Tracef(\"reading message from topic '%s'\", s.Config.Topic)\n\n\t\tm, err := s.Reader.ReadMessage(ctx)\n\t\tif err != nil {\n\t\t\tif errors.Is(err, io.EOF) {\n\t\t\t\treturn nil\n\t\t\t}\n\n\t\t\ts.logger.Errorln(fmt.Errorf(\"while reading %s message: %w\", s.GetName(), err))\n\n\t\t\tcontinue\n\t\t}\n\n\t\ts.logger.Tracef(\"got message: %s\", string(m.Value))","sourceCodeStart":5,"sourceCodeEnd":41,"githubUrl":"https://github.com/crowdsecurity/crowdsec/blob/909b5157986a2b2c2163300fdaef5ed01289f7d2/pkg/acquisition/modules/kafka/run.go#L5-L41","documentation":"Returned by the kafka Source.ReadMessage when Reader.SetOffset(kafka.LastOffset) fails. SetOffset is only called when no `group_id` is configured (consumer-group readers cannot call SetOffset in kafka-go), so this happens in partition/single-partition mode when positioning the reader at the last offset fails.","triggerScenarios":"ReadMessage is called on a kafka datasource without `group_id`; s.Reader.SetOffset(kafka.LastOffset) returns an error — typically because SetOffset was called after the reader already started fetching (kafka-go: 'SetOffset is not supported when GroupID is set' or 'cannot call SetOffset on an active reader'), or the client is already closed.","commonSituations":"Reader reused after Run/ReadMessage already began; concurrent calls; reader closed by shutdown then restarted; kafka-go version where offset setting rules changed.","solutions":["Call SetOffset only before the first ReadMessage/Run on the reader, exactly as ReadMessage does at startup","If you hit 'cannot call SetOffset on an active reader', recreate the kafka.Reader instead of reusing it","Ensure a group_id is actually empty in the yaml if you expect SetOffset to be called at all","Upgrade/align kafka-go version if error semantics differ from expected"],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"if err := src.ReadMessage(ctx, out); err != nil {\n    if strings.Contains(err.Error(), \"setting offset\") {\n        logger.Errorf(\"recreate the kafka.Reader before retrying SetOffset: %v\", err)\n    }\n    return err\n}","preventionTips":["Never call SetOffset after reads have started; set StartOffset in ReaderConfig instead","Recreate the reader instead of reusing one that errored","Only expect SetOffset behavior in non-group_id mode"],"tags":["kafka","offset","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-14T05:17:10.506Z"}