crowdsecurity/crowdsec · error

consumer %s is not active after %d tries

Error message

consumer %s is not active after %d tries

What it means

Returned by WaitForConsumerRegistration when the consumer never reaches ACTIVE status within `MaxRetries` polls (each 200ms*(i+1) apart). Kinesis EFO consumers typically take seconds to activate; this error means the polling budget was exhausted.

Source

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

	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)
		}

		if describeOutput.ConsumerDescription.ConsumerStatus == "ACTIVE" {
			s.logger.Debugf("Consumer %s is active", consumerARN)
			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)
	}

View on GitHub (pinned to 909b515798)

Solutions

1. Increase `max_retries` in the kinesis acquisition config (e.g. from a few to 15-30) to extend the ~200ms-growing backoff window.
2. Just restart the acquisition — the consumer persists and may already be ACTIVE; DescribeStreamConsumer on an existing consumer name returns its state.
3. Check for account-level consumer limits (up to 20 per stream) — if exhausted, deregister stale consumers.
4. For localstack dev environments, verify the version supports EFO consumer APIs properly.

Example fix

// before
max_retries: 3
// after
max_retries: 20

// Check current consumer state and count before starting EFO
consumers, err := client.ListStreamConsumers(ctx, &kinesis.ListStreamConsumersInput{StreamARN: aws.String(arn)})
if err != nil { return err }
if len(consumers.Consumers) >= 20 { return errors.New("stream at EFO consumer limit") }

if err := waitForActive(ctx, arn); err != nil {
    select {
    case <-ctx.Done():
        return ctx.Err()
    case <-time.After(2 * time.Second):
        return waitForActive(ctx, arn) // one extra round before giving up
    }
}

When it happens

Trigger: RegisterConsumer succeeded, but repeated DescribeStreamConsumer calls keep returning CREATING until maxTries is hit; RegisterConsumer then wraps it as "timeout while waiting for consumer to be active".

Common situations: Consumer activation under heavy stream load taking longer than the default budget; very small `max_retries` in config; localstack/test environments where activation is slower or mocked inconsistently.

Understand the failure class

Background: Request timed out: what client-side request timeouts mean across libraries (Request timed out, TIMED_OUT, APITimeoutError) — this error's family across 39 libraries.

Related errors

• timeout while waiting for consumer to be active: %w
• consumer %s is not deregistered after %d tries
• cannot wait for consumer deregistration: %w
• cannot register stream consumer: %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/e76ca55415e526ad. Report an issue: GitHub.