crowdsecurity/crowdsec · error

while setting offset for reader on topic '%s': %w

Error message

while setting offset for reader on topic '%s': %w

What it means

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.

Source

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

	"errors"
	"fmt"
	"io"

	"github.com/prometheus/client_golang/prometheus"
	"github.com/segmentio/kafka-go"
	"gopkg.in/tomb.v2"

	"github.com/crowdsecurity/go-cs-lib/trace"

	"github.com/crowdsecurity/crowdsec/pkg/metrics"
	"github.com/crowdsecurity/crowdsec/pkg/pipeline"
)

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))

View on GitHub (pinned to 909b515798)

Solutions

1. Call SetOffset only before the first ReadMessage/Run on the reader, exactly as ReadMessage does at startup
2. If you hit 'cannot call SetOffset on an active reader', recreate the kafka.Reader instead of reusing it
3. Ensure a group_id is actually empty in the yaml if you expect SetOffset to be called at all
4. Upgrade/align kafka-go version if error semantics differ from expected

if err := src.ReadMessage(ctx, out); err != nil {
    if strings.Contains(err.Error(), "setting offset") {
        logger.Errorf("recreate the kafka.Reader before retrying SetOffset: %v", err)
    }
    return err
}

When it happens

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

Common situations: Reader reused after Run/ReadMessage already began; concurrent calls; reader closed by shutdown then restarted; kafka-go version where offset setting rules changed.

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 closing %s 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/ed59d4eb988231ea. Report an issue: GitHub.