crowdsecurity/crowdsec · error

cannot deregister consumer: %w

Error message

cannot deregister consumer: %w

What it means

DeregisterConsumer failed at startup of enhanced fan-out mode. EnhancedRead always deregisters any pre-existing consumer with the same name before re-registering, to avoid ResourceInUse exceptions; this error wraps whatever AWS error (permissions, throttling, not found) that cleanup call returned.

Source

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

	return nil
}

func (s *Source) EnhancedRead(ctx context.Context, out chan pipeline.Event, t *tomb.Tomb) error {
	parsedARN, err := arn.Parse(s.Config.StreamARN)
	if err != nil {
		return fmt.Errorf("cannot parse stream ARN: %w", err)
	}

	if !strings.HasPrefix(parsedARN.Resource, "stream/") {
		return fmt.Errorf("resource part of stream ARN %s does not start with stream/", s.Config.StreamARN)
	}

	s.logger = s.logger.WithField("stream", parsedARN.Resource[7:])
	s.logger.Info("starting kinesis acquisition with enhanced fan-out")

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

	streamConsumer, err := s.RegisterConsumer(ctx)
	if err != nil {
		return fmt.Errorf("cannot register consumer: %w", err)
	}

	for {
		s.shardReaderTomb = &tomb.Tomb{}

		err = s.SubscribeToShards(ctx, parsedARN, streamConsumer, out)
		if err != nil {
			return fmt.Errorf("cannot subscribe to shards: %w", err)
		}

		select {
		case <-t.Dying():
			s.logger.Infof("Kinesis source is dying")

View on GitHub (pinned to 909b515798)

Solutions

1. Check IAM permissions for kinesis:DeregisterStreamConsumer on the consumer/stream.
2. If the consumer no longer exists, make DeregisterConsumer tolerate ResourceNotFoundException and treat it as success.
3. Verify region/account consistency of the persisted consumer ARN.
4. Retry on throttling with backoff; then rerun the datasource.

Example fix

// before: any error aborts startup
err = s.DeregisterConsumer(ctx)
// after: tolerate already-gone consumers
var rnfe *kinTypes.ResourceNotFoundException
if err != nil && !errors.As(err, &rnfe) {
	return fmt.Errorf("cannot deregister consumer: %w", err)
}

var rnfe *kinTypes.ResourceNotFoundException
if err := deregister(ctx); err != nil && !errors.As(err, &rnfe) {
	return fmt.Errorf("cannot deregister consumer: %w", err)
}

When it happens

Trigger: EnhancedRead calling s.DeregisterConsumer(ctx) when the IAM principal lacks kinesis:DeregisterStreamConsumer; the consumer ARN from a previous run is in another account/region; AWS returns LimitExceededException or a throttling error during the deregister call; the stored consumer ARN is stale (consumer was already removed manually).

Common situations: IAM policy hardened after initial setup; leftover consumer from a crashed run was deleted by hand, so deregister hits ResourceNotFound; account-level enhanced fan-out throttling; cross-account stream with a consumer ARN registered elsewhere.

Understand the failure class

Background: Permission denied / not authorized / 403 Forbidden: access-control rejections when the caller lacks the required role, grant, or ownership — this error's family across 18 libraries.

Related errors

• cannot list shards for enhanced_read: %w
• cannot subscribe to shard: %w
• cannot register consumer: %w
• cannot get shard iterator: %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/e6419d1342921e43. Report an issue: GitHub.