crowdsecurity/crowdsec · error

cannot deregister stream consumer: %w

Error message

cannot deregister stream consumer: %w

What it means

Returned by DeregisterConsumer when the DeregisterStreamConsumer API call fails with an error that is not a ResourceNotFoundException (that case is treated as already-deregistered and returns nil). Wraps the raw SDK error from EnhancedRead's consumer cleanup path.

Source

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

	}

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

func (s *Source) DeregisterConsumer(ctx context.Context) error {
	s.logger.Debugf("Deregistering consumer %s if it exists", s.Config.ConsumerName)
	_, err := s.kClient.DeregisterStreamConsumer(ctx, &kinesis.DeregisterStreamConsumerInput{
			ConsumerName: aws.String(s.Config.ConsumerName),
			StreamARN:    aws.String(s.Config.StreamARN),
		})

	var resourceNotFoundErr *kinTypes.ResourceNotFoundException
	if errors.As(err, &resourceNotFoundErr) {
		return nil
	}

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

	err = s.WaitForConsumerDeregistration(ctx, s.Config.ConsumerName, s.Config.StreamARN)
	if err != nil {
		return fmt.Errorf("cannot wait for consumer deregistration: %w", err)
	}

	return nil
}

func (s *Source) WaitForConsumerRegistration(ctx context.Context, consumerARN string) error {
	maxTries := s.Config.MaxRetries
	for i := range maxTries {
		describeOutput, err := s.kClient.DescribeStreamConsumer(ctx, &kinesis.DescribeStreamConsumerInput{
				ConsumerARN: aws.String(consumerARN),
			})
		if err != nil {
			return fmt.Errorf("cannot describe stream consumer: %w", err)

View on GitHub (pinned to 909b515798)

Solutions

1. Inspect the wrapped cause and check the consumer name and stream ARN in the acquisition config for typos and region correctness.
2. Add `kinesis:DeregisterStreamConsumer` to the IAM policy of the credentials in use.
3. If the error is InvalidArgumentException, ensure consumer_name is 1-128 chars of alphanumerics plus ._- and stream_arn is a full valid ARN.
4. Retry on transient network errors; deregistration is idempotent thanks to the ResourceNotFound short-circuit.

Example fix

// before
{"stream_arn": "my-stream"}
// after
{"stream_arn": "arn:aws:kinesis:us-east-1:123456789012:stream/my-stream"}

// Pre-flight: consumer name constraints and ARN shape
var re = regexp.MustCompile(`^[a-zA-Z0-9_.-]{1,128}$`)
if !re.MatchString(consumerName) { return errors.New("invalid consumer_name") }
if !strings.HasPrefix(streamARN, "arn:aws:kinesis:") { return errors.New("invalid stream_arn") }

var nf *kinTypes.ResourceNotFoundException
if errors.As(err, &nf) { return nil }
var invalid *kinTypes.InvalidArgumentException
if errors.As(err, &invalid) { /* fix config: bad ARN or name */ }

_, err := client.DeregisterStreamConsumer(ctx, in)
if err != nil {
    var nf *kinTypes.ResourceNotFoundException
    if errors.As(err, &nf) { return nil }
    return fmt.Errorf("cannot deregister stream consumer: %w", err)
}

When it happens

Trigger: EnhancedRead calls DeregisterConsumer at startup/shutdown and DeregisterStreamConsumer returns e.g. InvalidArgumentException (malformed StreamARN/ConsumerName), AccessDeniedException, or a network error.

Common situations: IAM policy missing kinesis:DeregisterStreamConsumer; stream ARN typo or wrong region; consumer name longer than 128 characters or containing invalid characters.

Related errors

• cannot register stream consumer: %w
• stream_name is mandatory when use_enhanced_fanout is false
• stream_arn is mandatory when use_enhanced_fanout is true
• consumer_name is mandatory when use_enhanced_fanout is true
• stream_arn and stream_name are mutually exclusive

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