{"record":{"id":"a3bbdb16cb23b8a1","repo":"crowdsecurity/crowdsec","slug":"cannot-subscribe-to-shard-w","errorCode":null,"errorMessage":"cannot subscribe to shard: %w","messagePattern":"cannot subscribe to shard: %w","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"pkg/acquisition/modules/kinesis/run.go","lineNumber":262,"sourceCode":"\nfunc (s *Source) SubscribeToShards(ctx context.Context, arn arn.ARN, streamConsumer *kinesis.RegisterStreamConsumerOutput, out chan pipeline.Event) error {\n\tshards, err := s.kClient.ListShards(ctx, &kinesis.ListShardsInput{\n\t\t\tStreamName: aws.String(arn.Resource[7:]),\n\t\t})\n\tif err != nil {\n\t\treturn fmt.Errorf(\"cannot list shards for enhanced_read: %w\", err)\n\t}\n\n\tfor _, shard := range shards.Shards {\n\t\tshardID := *shard.ShardId\n\n\t\tr, err := s.kClient.SubscribeToShard(ctx, &kinesis.SubscribeToShardInput{\n\t\t\t\tShardId:          aws.String(shardID),\n\t\t\t\tStartingPosition: &kinTypes.StartingPosition{Type: kinTypes.ShardIteratorTypeLatest},\n\t\t\t\tConsumerARN:      streamConsumer.Consumer.ConsumerARN,\n\t\t\t})\n\t\tif err != nil {\n\t\t\treturn fmt.Errorf(\"cannot subscribe to shard: %w\", err)\n\t\t}\n\n\t\ts.shardReaderTomb.Go(func() error {\n\t\t\treturn s.ReadFromSubscription(r.GetStream().Reader, out, shardID, arn.Resource[7:])\n\t\t})\n\t}\n\n\treturn nil\n}\n\nfunc (s *Source) EnhancedRead(ctx context.Context, out chan pipeline.Event, t *tomb.Tomb) error {\n\tparsedARN, err := arn.Parse(s.Config.StreamARN)\n\tif err != nil {\n\t\treturn fmt.Errorf(\"cannot parse stream ARN: %w\", err)\n\t}\n\n\tif !strings.HasPrefix(parsedARN.Resource, \"stream/\") {\n\t\treturn fmt.Errorf(\"resource part of stream ARN %s does not start with stream/\", s.Config.StreamARN)","sourceCodeStart":244,"sourceCodeEnd":280,"githubUrl":"https://github.com/crowdsecurity/crowdsec/blob/909b5157986a2b2c2163300fdaef5ed01289f7d2/pkg/acquisition/modules/kinesis/run.go#L244-L280","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Confirm the consumer is registered: aws kinesis list-stream-consumers --stream-arn <arn>; re-register if missing.","Handle ResourceNotFound for stale shards: refresh the shard list and retry (the EnhancedRead loop already resubscribes on clean tomb death).","Check enhanced fan-out limits (consumers per stream); fall back to the classic poll mode by disabling enhanced fan-out in the config.","Grant kinesis:SubscribeToShard on the consumer and stream in the IAM policy.","Ensure only one process registers/deregisters this consumer name to avoid races."],"exampleFix":"// before: blind retry on any shard error\nreturn fmt.Errorf(\"cannot subscribe to shard: %w\", err)\n// after: tolerate vanished shards, fail on the rest\nvar rnfe *kinTypes.ResourceNotFoundException\nif errors.As(err, &rnfe) {\n\tlogger.Warnf(\"shard %s gone (resharding?), skipping\", shardID)\n\tcontinue\n}\nreturn fmt.Errorf(\"cannot subscribe to shard: %w\", err)","handlingStrategy":"retry","validationCode":"consumers, err := client.ListStreamConsumers(ctx, &kinesis.ListStreamConsumersInput{StreamARN: aws.String(streamARN)})\n// verify consumerName is registered before subscribing; count must stay under the account limit","typeGuard":null,"tryCatchPattern":"var rie *kinTypes.ResourceInUseException\nif err := subscribe(); err != nil {\n\tswitch {\n\tcase errors.As(err, &rie):\n\t\t// consumer in use: deregister + retry\n\tcase errors.As(err, &limitErr):\n\t\t// fall back to classic polling\n\tdefault:\n\t\treturn err\n\t}\n}","preventionTips":["Use a unique consumer name per crowdsec instance.","Keep enhanced fan-out consumer count within account limits; delete unused consumers.","Re-run SubscribeToShards periodically (the code already does) and tolerate reshard races.","Grant kinesis:SubscribeToShard in IAM."],"tags":["aws","kinesis","iam","streaming"],"backgroundTag":"api-request-failed","analyzedSha":"909b5157986a2b2c2163300fdaef5ed01289f7d2","analyzedAt":"2026-09-06T12:27:26.012Z","contentChangedAt":"2026-09-06T12:27:26.012Z","schemaVersion":2},"datasetVersion":"2026-09-14T05:17:10.506Z"}