crowdsecurity/crowdsec · error

cannot register consumer: %w

Error message

cannot register consumer: %w

What it means

RegisterStreamConsumer failed while EnhancedRead set up the enhanced fan-out consumer for the stream. Registering a consumer can fail due to duplicate names (ResourceInUse), enhanced fan-out quotas, missing IAM permission, or invalid stream state. Without a registered consumer, no SubscribeToShard call can succeed, so startup aborts.

Source

Thrown at pkg/acquisition/modules/kinesis/run.go:293

	if err != nil {
		return fmt.Errorf("cannot parse stream ARN: %w", err)
	}

	if !strings.HasPrefix(parsedARN.Resource, "stream/") {
		return fmt.Errorf("resource part of stream ARN %s does not start with stream/", s.Config.StreamARN)
	}

	s.logger = s.logger.WithField("stream", parsedARN.Resource[7:])
	s.logger.Info("starting kinesis acquisition with enhanced fan-out")

	err = s.DeregisterConsumer(ctx)
	if err != nil {
		return fmt.Errorf("cannot deregister consumer: %w", err)
	}

	streamConsumer, err := s.RegisterConsumer(ctx)
	if err != nil {
		return fmt.Errorf("cannot register consumer: %w", err)
	}

	for {
		s.shardReaderTomb = &tomb.Tomb{}

		err = s.SubscribeToShards(ctx, parsedARN, streamConsumer, out)
		if err != nil {
			return fmt.Errorf("cannot subscribe to shards: %w", err)
		}

		select {
		case <-t.Dying():
			s.logger.Infof("Kinesis source is dying")
			s.shardReaderTomb.Kill(nil)
			_ = s.shardReaderTomb.Wait() // we don't care about the error as we kill the tomb ourselves

			err = s.DeregisterConsumer(ctx)
			if err != nil {

View on GitHub (pinned to 909b515798)

Solutions

1. List existing consumers (aws kinesis list-stream-consumers --stream-arn <arn>) and remove the stale one with deregister-stream-consumer.
2. Use a unique consumer name per crowdsec instance in the config.
3. Add kinesis:RegisterStreamConsumer (and SubscribeToShard) to the IAM policy.
4. Check the stream is ACTIVE; wait if it's being rescaled.
5. If hitting quota limits, reduce consumers or fall back to classic shard polling.

Example fix

// before: same consumer name for both instances
consumer_name: crowdsec
// after: instance-specific name
consumer_name: crowdsec-node1

// pre-check registration state
resp, err := client.ListStreamConsumers(ctx, &kinesis.ListStreamConsumersInput{StreamARN: aws.String(streamARN)})
for _, c := range resp.Consumers {
	if *c.ConsumerName == consumerName && *c.ConsumerStatus == "ACTIVE" {
		// already registered: skip RegisterStreamConsumer
	}
}

var riue *kinTypes.ResourceInUseException
if err := register(); err != nil {
	if errors.As(err, &riue) {
		// existing consumer OK: reuse instead of failing
	} else if errors.As(err, &limitErr) {
		// wait/backoff, quota full
	}
}

When it happens

Trigger: EnhancedRead calling s.RegisterConsumer(ctx) when a consumer with the same name is already ACTIVE (ResourceInUseException), the account hit the 20-consumers-per-stream limit (LimitExceededException), IAM lacks kinesis:RegisterStreamConsumer, or the stream is not ACTIVE.

Common situations: Two crowdsec instances sharing one consumer name; a previous crash left a consumer pending-registration; account recently enabled enhanced fan-out and hit limits; policy update removed RegisterStreamConsumer.

Understand the failure class

Background: "API request failed": what wrapped HTTP errors from external APIs mean and how to find the real cause — this error's family across 29 libraries.

Related errors

• cannot list shards for enhanced_read: %w
• cannot subscribe to shard: %w
• cannot deregister consumer: %w
• cannot get shard iterator: %w
• stream_name is mandatory when use_enhanced_fanout is false

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