{"record":{"id":"68c02b2792fa6999","repo":"crowdsecurity/crowdsec","slug":"cannot-register-consumer-w","errorCode":null,"errorMessage":"cannot register consumer: %w","messagePattern":"cannot register consumer: %w","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"pkg/acquisition/modules/kinesis/run.go","lineNumber":293,"sourceCode":"\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)\n\t}\n\n\ts.logger = s.logger.WithField(\"stream\", parsedARN.Resource[7:])\n\ts.logger.Info(\"starting kinesis acquisition with enhanced fan-out\")\n\n\terr = s.DeregisterConsumer(ctx)\n\tif err != nil {\n\t\treturn fmt.Errorf(\"cannot deregister consumer: %w\", err)\n\t}\n\n\tstreamConsumer, err := s.RegisterConsumer(ctx)\n\tif err != nil {\n\t\treturn fmt.Errorf(\"cannot register consumer: %w\", err)\n\t}\n\n\tfor {\n\t\ts.shardReaderTomb = &tomb.Tomb{}\n\n\t\terr = s.SubscribeToShards(ctx, parsedARN, streamConsumer, out)\n\t\tif err != nil {\n\t\t\treturn fmt.Errorf(\"cannot subscribe to shards: %w\", err)\n\t\t}\n\n\t\tselect {\n\t\tcase <-t.Dying():\n\t\t\ts.logger.Infof(\"Kinesis source is dying\")\n\t\t\ts.shardReaderTomb.Kill(nil)\n\t\t\t_ = s.shardReaderTomb.Wait() // we don't care about the error as we kill the tomb ourselves\n\n\t\t\terr = s.DeregisterConsumer(ctx)\n\t\t\tif err != nil {","sourceCodeStart":275,"sourceCodeEnd":311,"githubUrl":"https://github.com/crowdsecurity/crowdsec/blob/909b5157986a2b2c2163300fdaef5ed01289f7d2/pkg/acquisition/modules/kinesis/run.go#L275-L311","documentation":"RegisterStreamConsumer failed while EnhancedRead set up the enhanced fan-out consumer for the stream. Registering a consumer can fail due to duplicate names (ResourceInUse), enhanced fan-out quotas, missing IAM permission, or invalid stream state. Without a registered consumer, no SubscribeToShard call can succeed, so startup aborts.","triggerScenarios":"EnhancedRead calling s.RegisterConsumer(ctx) when a consumer with the same name is already ACTIVE (ResourceInUseException), the account hit the 20-consumers-per-stream limit (LimitExceededException), IAM lacks kinesis:RegisterStreamConsumer, or the stream is not ACTIVE.","commonSituations":"Two crowdsec instances sharing one consumer name; a previous crash left a consumer pending-registration; account recently enabled enhanced fan-out and hit limits; policy update removed RegisterStreamConsumer.","solutions":["List existing consumers (aws kinesis list-stream-consumers --stream-arn <arn>) and remove the stale one with deregister-stream-consumer.","Use a unique consumer name per crowdsec instance in the config.","Add kinesis:RegisterStreamConsumer (and SubscribeToShard) to the IAM policy.","Check the stream is ACTIVE; wait if it's being rescaled.","If hitting quota limits, reduce consumers or fall back to classic shard polling."],"exampleFix":"// before: same consumer name for both instances\nconsumer_name: crowdsec\n// after: instance-specific name\nconsumer_name: crowdsec-node1","handlingStrategy":"retry","validationCode":"// pre-check registration state\nresp, err := client.ListStreamConsumers(ctx, &kinesis.ListStreamConsumersInput{StreamARN: aws.String(streamARN)})\nfor _, c := range resp.Consumers {\n\tif *c.ConsumerName == consumerName && *c.ConsumerStatus == \"ACTIVE\" {\n\t\t// already registered: skip RegisterStreamConsumer\n\t}\n}","typeGuard":null,"tryCatchPattern":"var riue *kinTypes.ResourceInUseException\nif err := register(); err != nil {\n\tif errors.As(err, &riue) {\n\t\t// existing consumer OK: reuse instead of failing\n\t} else if errors.As(err, &limitErr) {\n\t\t// wait/backoff, quota full\n\t}\n}","preventionTips":["One consumer name per instance; include hostname in the name.","Monitor consumer count per stream.","Keep stream ACTIVE before starting the datasource.","IAM: RegisterStreamConsumer + SubscribeToShard."],"tags":["aws","kinesis","iam","quota"],"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"}