crowdsecurity/crowdsec · warning

while closing %s reader on topic '%s': %w

Error message

while closing %s reader on topic '%s': %w

What it means

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.

Source

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

		evt := pipeline.MakeEvent(s.Config.UseTimeMachine, pipeline.LOG, true)
		evt.Line = l

		out <- evt
	}
}

func (s *Source) RunReader(ctx context.Context, out chan pipeline.Event, t *tomb.Tomb) error {
	s.logger.Debugf("starting %s datasource reader goroutine with configuration %+v", s.GetName(), s.Config)
	t.Go(func() error {
		return s.ReadMessage(ctx, out)
	})

	<-t.Dying()

	s.logger.Infof("%s datasource topic %s stopping", s.GetName(), s.Config.Topic)

	if err := s.Reader.Close(); err != nil {
		return fmt.Errorf("while closing %s reader on topic '%s': %w", s.GetName(), s.Config.Topic, err)
	}

	return nil
}

func (s *Source) StreamingAcquisition(ctx context.Context, out chan pipeline.Event, t *tomb.Tomb) error {
	s.logger.Infof("start reader on brokers '%+v' with topic '%s'", s.Config.Brokers, s.Config.Topic)

	t.Go(func() error {
		defer trace.ReportPanic()
		return s.RunReader(ctx, out, t)
	})

	return nil
}

View on GitHub (pinned to 909b515798)

Solutions

1. Ensure no concurrent ReadMessage call is active when Close is invoked (cancel/drain the read loop first)
2. Check broker availability during shutdown; retry the reload once the broker is back
3. Inspect the wrapped %w error: 'reader was closed' indicates double Close — fix the lifecycle to close once
4. As a last resort this is usually harmless at process exit; treat as noise only after confirming connections are cleaned up

if err := reader.Close(); err != nil {
    if strings.Contains(err.Error(), "closed") {
        return nil // double-close, ignore
    }
    logger.Warnf("kafka reader close during shutdown: %v", err)
}

When it happens

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

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

Understand the failure class

Background: "Invalid state transition" errors: "status must be X, actually Y", "already rejected/charging/uninstalled", "cannot ... while running" — what they mean when a library rejects your call — this error's family across 31 libraries.

Related errors

• while validating reader configuration: %w
• while setting offset for reader on topic '%s': %w
• cannot create %s dialer: %w
• cannote create %s reader: %w
• cannot create %s reader

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