{"record":{"id":"0cad0972963996ae","repo":"crowdsecurity/crowdsec","slug":"while-reading-s-message-w","errorCode":null,"errorMessage":"while reading %s message: %w","messagePattern":"while reading (.+?) message: %w","errorType":"console","errorClass":null,"httpStatus":null,"severity":"error","filePath":"pkg/acquisition/modules/kafka/run.go","lineNumber":36,"sourceCode":"\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))\n\t\tl := pipeline.Line{\n\t\t\tRaw:     string(m.Value),\n\t\t\tLabels:  s.Config.Labels,\n\t\t\tTime:    m.Time.UTC(),\n\t\t\tSrc:     s.Config.Topic,\n\t\t\tProcess: true,\n\t\t\tModule:  s.GetName(),\n\t\t}\n\t\ts.logger.Tracef(\"line with message read from topic '%s': %+v\", s.Config.Topic, l)\n\n\t\tif s.metricsLevel != metrics.AcquisitionMetricsLevelNone {\n\t\t\tmetrics.KafkaDataSourceLinesRead.With(prometheus.Labels{\"topic\": s.Config.Topic, \"datasource_type\": ModuleName, \"acquis_type\": l.Labels[\"type\"]}).Inc()\n\t\t}","sourceCodeStart":18,"sourceCodeEnd":54,"githubUrl":"https://github.com/crowdsecurity/crowdsec/blob/909b5157986a2b2c2163300fdaef5ed01289f7d2/pkg/acquisition/modules/kafka/run.go#L18-L54","documentation":"Logged (via s.logger.Errorln, then the loop continues) when Reader.ReadMessage(ctx) fails with a non-EOF error while consuming from the kafka topic. EOF is treated as a clean stop; everything else is reported with the source name and read retried indefinitely. This is a runtime error, not a configuration one — the datasource keeps running and retrying.","triggerScenarios":"ReadMessage's loop calls s.Reader.ReadMessage(ctx) and gets a non-EOF error: broker unreachable, connection reset, context canceled during shutdown, topic/partition leadership changes, or offsets out of range in non-group mode.","commonSituations":"Kafka broker down or restarted; network partition between crowdsec and the broker; topic deleted/recreated; ctx cancellation when crowdsec is reloading (expect bursts of these during shutdown); TLS handshake failures surfacing at read time.","solutions":["Check kafka broker reachability (telnet/nc to broker:9092) and broker logs","Look at the wrapped %w error to distinguish context.Canceled (expected during reload/shutdown) from real connection errors","Verify the topic exists and partitions have data: `kafka-topics.sh --describe`","If TLS, verify client certs/CA on the dialer are still valid and unexpired"],"exampleFix":null,"handlingStrategy":"retry","validationCode":"// pre-flight: broker reachable\nconn, err := net.DialTimeout(\"tcp\", brokerAddr, 5*time.Second)\nif err != nil { return fmt.Errorf(\"kafka broker unreachable: %w\", err) }\nconn.Close()","typeGuard":null,"tryCatchPattern":"if err := src.ReadMessage(ctx, out); err != nil {\n    if errors.Is(err, context.Canceled) {\n        return nil // expected during shutdown\n    }\n    // treat as transient: the datasource loop already retries; add backoff/alerting\n    return err\n}","preventionTips":["Monitor broker liveness and alert before crowdsec log-spams read errors","Distinguish context.Canceled (reload) from connection errors in log triage","Use group_id mode for resilient multi-broker consumption"],"tags":["kafka","runtime","network","kafka-go"],"backgroundTag":"connection-refused","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"}