crowdsecurity/crowdsec · error

cannot describe stream consumer: %w

Error message

cannot describe stream consumer: %w

What it means

Returned by WaitForConsumerDeregistration when a DescribeStreamConsumer call fails with a non-ResourceNotFound error while polling for a consumer to leave DELETING state. After DeregisterStreamConsumer, the source polls the consumer; a describe failure other than ResourceNotFound aborts the wait with this wrapped error.

Source

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

	return subscriptionRecord.LogEvents, nil
}

func (s *Source) WaitForConsumerDeregistration(ctx context.Context, consumerName string, streamARN string) error {
	maxTries := s.Config.MaxRetries
	for i := range maxTries {
		_, err := s.kClient.DescribeStreamConsumer(ctx, &kinesis.DescribeStreamConsumerInput{
				ConsumerName: aws.String(consumerName),
				StreamARN:    aws.String(streamARN),
			})

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

		if err != nil {
			s.logger.Errorf("Error while waiting for consumer deregistration: %s", err)
			return fmt.Errorf("cannot describe stream consumer: %w", err)
		}

		time.Sleep(time.Millisecond * 200 * time.Duration(i+1))
	}

	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

View on GitHub (pinned to 909b515798)

Solutions

1. Check the wrapped cause: if throttled, reduce polling frequency or increase Config.MaxRetries with backoff.
2. Grant the IAM principal `kinesis:DescribeStreamConsumer` and `kinesis:DeregisterStreamConsumer` on the stream/consumer ARN.
3. Verify `stream_arn` in the acquisition config still refers to an existing stream.
4. Check network/proxy reachability to the Kinesis endpoint.

Example fix

// before
{"stream_arn": "arn:aws:kinesis:us-east-1:123:stream/old-stream"}
// after
{"stream_arn": "arn:aws:kinesis:us-east-1:123:stream/current-stream"}

// Ensure IAM actions exist before running
// kinesis:DescribeStreamConsumer on stream/consumer ARNs
aws iam simulate-principal-policy --policy-source-arn <role> --action-names kinesis:DescribeStreamConsumer

var nf *kinTypes.ResourceNotFoundException
if errors.As(err, &nf) { return nil } // consumer already gone — treat as success

_, err := s.kClient.DescribeStreamConsumer(ctx, input)
var nf *kinTypes.ResourceNotFoundException
if errors.As(err, &nf) { return nil }
var tle *kinTypes.LimitExceededException
if errors.As(err, &tle) { /* retry with backoff */ }

When it happens

Trigger: Calling DeregisterConsumer (e.g. during source shutdown in EnhancedRead teardown) and the DescribeStreamConsumer API returns an unexpected error: throttling, invalid ARN format, network failure, or expired credentials.

Common situations: AWS throttling (LimitExceededException) during shutdown bursts; IAM role lacking kinesis:DescribeStreamConsumer; stale StreamARN after the stream was deleted/recreated.

Related errors

• 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
• cannot create kinesis client: %w

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