crowdsecurity/crowdsec · error

cannot register stream consumer: %w

Error message

cannot register stream consumer: %w

What it means

RegisterConsumer calls kinesis RegisterStreamConsumer for an EFO (enhanced fan-out) consumer; the AWS API call failed (permissions, invalid ARN, stream state, throttling), so the consumer could not be registered and EnhancedRead cannot proceed.

Source

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

			return nil
		}

		time.Sleep(time.Millisecond * 200 * time.Duration(i+1))
		s.logger.Debugf("Waiting for consumer registration %d", i)
	}

	return fmt.Errorf("consumer %s is not active after %d tries", consumerARN, maxTries)
}

func (s *Source) RegisterConsumer(ctx context.Context) (*kinesis.RegisterStreamConsumerOutput, error) {
	s.logger.Debugf("Registering consumer %s", s.Config.ConsumerName)

	streamConsumer, err := s.kClient.RegisterStreamConsumer(ctx, &kinesis.RegisterStreamConsumerInput{
			ConsumerName: aws.String(s.Config.ConsumerName),
			StreamARN:    aws.String(s.Config.StreamARN),
		})
	if err != nil {
		return nil, fmt.Errorf("cannot register stream consumer: %w", err)
	}

	err = s.WaitForConsumerRegistration(ctx, *streamConsumer.Consumer.ConsumerARN)
	if err != nil {
		return nil, fmt.Errorf("timeout while waiting for consumer to be active: %w", err)
	}

	return streamConsumer, nil
}

func (s *Source) ParseAndPushRecords(records []kinTypes.Record, out chan pipeline.Event, logger *log.Entry, shardID string) {
	for _, record := range records {
		if s.Config.StreamARN != "" {
			if s.metricsLevel != metrics.AcquisitionMetricsLevelNone {
				metrics.KinesisDataSourceLinesReadShards.With(prometheus.Labels{"stream": s.Config.StreamARN, "shard": shardID}).Inc()
				metrics.KinesisDataSourceLinesRead.With(prometheus.Labels{"stream": s.Config.StreamARN, "datasource_type": ModuleName, "acquis_type": s.Config.Labels["type"]}).Inc()
			}
		} else {

View on GitHub (pinned to 909b515798)

Solutions

1. Verify stream ARN, region and consumer name in the DSN
2. Check IAM permissions for kinesis:RegisterStreamConsumer
3. Ensure the consumer name is valid (≤128 chars, alphanumerics plus _.=+@-)

Example fix

// IAM policy addition
{"Effect": "Allow", "Action": ["kinesis:RegisterStreamConsumer", "kinesis:DescribeStreamConsumer", "kinesis:DeregisterStreamConsumer"], "Resource": "*"}

// Pre-flight checks before RegisterStreamConsumer
_, err := client.DescribeStreamSummary(ctx, &kinesis.DescribeStreamSummaryInput{StreamName: aws.String(name)})
if err != nil { return fmt.Errorf("stream missing or undescribable: %w", err) }
var re = regexp.MustCompile(`^[a-zA-Z0-9_.-]{1,128}$`)
if !re.MatchString(consumerName) { return errors.New("invalid consumer_name") }

var nf *kinTypes.ResourceNotFoundException
if errors.As(err, &nf) { /* stream or consumer ARN wrong — fix config */ }
var lim *kinTypes.LimitExceededException
if errors.As(err, &lim) { /* 20-consumer cap: deregister stale ones */ }

out, err := client.RegisterStreamConsumer(ctx, in)
if err != nil {
    var exists *kinTypes.ResourceInUseException
    if errors.As(err, &exists) { /* consumer already registered — proceed to describe */ }
    return nil, fmt.Errorf("cannot register stream consumer: %w", err)
}

When it happens

Trigger: RegisterStreamConsumer returns ResourceNotFoundException (stream doesn't exist), LimitExceededException (20 consumers per stream max), InvalidArgumentException (bad consumer name/ARN), AccessDeniedException, or a throttling/network error.

Common situations: Stream deleted or ARN typo; already 20 EFO consumers registered on the stream; IAM policy missing kinesis:RegisterStreamConsumer; running against localstack without EFO support.

Understand the failure class

Background: 'Could not be found', 'does not exist', 'not found in database': the resource-not-found family when an ID, slug, key, or URI lookup comes back empty — this error's family across 20 libraries.

Related errors

• cannot deregister stream consumer: %w
• consumer %s is not active after %d tries
• timeout while waiting for consumer to be active: %w
• stream_name is mandatory when use_enhanced_fanout is false
• stream_arn is mandatory when use_enhanced_fanout is true

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