crowdsecurity/crowdsec · error

cannot subscribe to shard: %w

Error message

cannot subscribe to shard: %w

What it means

SubscribeToShard failed while attaching the enhanced fan-out consumer to an individual shard. The SDK returns this when the consumer ARN is wrong/unregistered, the shard ID no longer exists (resharding), or the enhanced fan-out quota is exceeded. It is raised inside SubscribeToShards' loop over shards and aborts startup of the remaining subscriptions.

Source

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

func (s *Source) SubscribeToShards(ctx context.Context, arn arn.ARN, streamConsumer *kinesis.RegisterStreamConsumerOutput, out chan pipeline.Event) error {
	shards, err := s.kClient.ListShards(ctx, &kinesis.ListShardsInput{
			StreamName: aws.String(arn.Resource[7:]),
		})
	if err != nil {
		return fmt.Errorf("cannot list shards for enhanced_read: %w", err)
	}

	for _, shard := range shards.Shards {
		shardID := *shard.ShardId

		r, err := s.kClient.SubscribeToShard(ctx, &kinesis.SubscribeToShardInput{
				ShardId:          aws.String(shardID),
				StartingPosition: &kinTypes.StartingPosition{Type: kinTypes.ShardIteratorTypeLatest},
				ConsumerARN:      streamConsumer.Consumer.ConsumerARN,
			})
		if err != nil {
			return fmt.Errorf("cannot subscribe to shard: %w", err)
		}

		s.shardReaderTomb.Go(func() error {
			return s.ReadFromSubscription(r.GetStream().Reader, out, shardID, arn.Resource[7:])
		})
	}

	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)

View on GitHub (pinned to 909b515798)

Solutions

1. Confirm the consumer is registered: aws kinesis list-stream-consumers --stream-arn <arn>; re-register if missing.
2. Handle ResourceNotFound for stale shards: refresh the shard list and retry (the EnhancedRead loop already resubscribes on clean tomb death).
3. Check enhanced fan-out limits (consumers per stream); fall back to the classic poll mode by disabling enhanced fan-out in the config.
4. Grant kinesis:SubscribeToShard on the consumer and stream in the IAM policy.
5. Ensure only one process registers/deregisters this consumer name to avoid races.

Example fix

// before: blind retry on any shard error
return fmt.Errorf("cannot subscribe to shard: %w", err)
// after: tolerate vanished shards, fail on the rest
var rnfe *kinTypes.ResourceNotFoundException
if errors.As(err, &rnfe) {
	logger.Warnf("shard %s gone (resharding?), skipping", shardID)
	continue
}
return fmt.Errorf("cannot subscribe to shard: %w", err)

consumers, err := client.ListStreamConsumers(ctx, &kinesis.ListStreamConsumersInput{StreamARN: aws.String(streamARN)})
// verify consumerName is registered before subscribing; count must stay under the account limit

var rie *kinTypes.ResourceInUseException
if err := subscribe(); err != nil {
	switch {
	case errors.As(err, &rie):
		// consumer in use: deregister + retry
	case errors.As(err, &limitErr):
		// fall back to classic polling
	default:
		return err
	}
}

When it happens

Trigger: Calling s.kClient.SubscribeToShard with a ConsumerARN that is not registered or was deregistered; the shard was merged/closed between ListShards and SubscribeToShard; ResourceNotFoundException for the stream or consumer; LimitsExceededException when the account hit 20 consumers per stream / enhanced throughput limits; the consumer lacks kinesis:SubscribeToShard permission.

Common situations: A concurrent crowdsec instance or a manual run deregistered the consumer; stream was rescaled (resharding) right at startup; consumer name conflicts because DeregisterConsumer raced with another process; IAM policy grants stream access but not enhanced fan-out APIs.

Understand the failure class

Background: "API request failed": what wrapped HTTP errors from external APIs mean and how to find the real cause — this error's family across 29 libraries.

Related errors

• cannot get shard iterator: %w
• cannot list shards for enhanced_read: %w
• cannot deregister consumer: %w
• cannot register consumer: %w
• cannot subscribe to shards: %w

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