crowdsecurity/crowdsec · error

cannot register stream consumer

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.

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
• consumer is not active after tries
• timeout while waiting for consumer to be active
• cannot close kinesis subscribed shard reader
• cannot create kinesis client

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

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