crowdsecurity/crowdsec · error

while reading %s message: %w

Error message

while reading %s message: %w

What it means

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.

Source

Thrown at pkg/acquisition/modules/kafka/run.go:36

func (s *Source) ReadMessage(ctx context.Context, out chan pipeline.Event) error {
	if s.Config.GroupID == "" {
		err := s.Reader.SetOffset(kafka.LastOffset)
		if err != nil {
			return fmt.Errorf("while setting offset for reader on topic '%s': %w", s.Config.Topic, err)
		}
	}

	for {
		s.logger.Tracef("reading message from topic '%s'", s.Config.Topic)

		m, err := s.Reader.ReadMessage(ctx)
		if err != nil {
			if errors.Is(err, io.EOF) {
				return nil
			}

			s.logger.Errorln(fmt.Errorf("while reading %s message: %w", s.GetName(), err))

			continue
		}

		s.logger.Tracef("got message: %s", string(m.Value))
		l := pipeline.Line{
			Raw:     string(m.Value),
			Labels:  s.Config.Labels,
			Time:    m.Time.UTC(),
			Src:     s.Config.Topic,
			Process: true,
			Module:  s.GetName(),
		}
		s.logger.Tracef("line with message read from topic '%s': %+v", s.Config.Topic, l)

		if s.metricsLevel != metrics.AcquisitionMetricsLevelNone {
			metrics.KafkaDataSourceLinesRead.With(prometheus.Labels{"topic": s.Config.Topic, "datasource_type": ModuleName, "acquis_type": l.Labels["type"]}).Inc()
		}

View on GitHub (pinned to 909b515798)

Solutions

1. Check kafka broker reachability (telnet/nc to broker:9092) and broker logs
2. Look at the wrapped %w error to distinguish context.Canceled (expected during reload/shutdown) from real connection errors
3. Verify the topic exists and partitions have data: `kafka-topics.sh --describe`
4. If TLS, verify client certs/CA on the dialer are still valid and unexpired

// pre-flight: broker reachable
conn, err := net.DialTimeout("tcp", brokerAddr, 5*time.Second)
if err != nil { return fmt.Errorf("kafka broker unreachable: %w", err) }
conn.Close()

if err := src.ReadMessage(ctx, out); err != nil {
    if errors.Is(err, context.Canceled) {
        return nil // expected during shutdown
    }
    // treat as transient: the datasource loop already retries; add backoff/alerting
    return err
}

When it happens

Trigger: 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.

Common situations: 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.

Understand the failure class

Background: ECONNREFUSED and "connection refused" / "could not connect to server" errors: what they mean and how to fix them — this error's family across 44 libraries.

Related errors

• while validating reader configuration: %w
• while setting offset for reader on topic '%s': %w
• while closing %s reader on topic '%s': %w
• cannot specify both group_id and partition
• error connecting to websocket

AI-assisted analysis of crowdsecurity/crowdsec@909b515798 (2026-09-06). Data as JSON: /api/errors/0cad0972963996ae. Report an issue: GitHub.