{"record":{"id":"e6419d1342921e43","repo":"crowdsecurity/crowdsec","slug":"cannot-deregister-consumer-w","errorCode":null,"errorMessage":"cannot deregister consumer: %w","messagePattern":"cannot deregister consumer: %w","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"pkg/acquisition/modules/kinesis/run.go","lineNumber":288,"sourceCode":"\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)\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\")","sourceCodeStart":270,"sourceCodeEnd":306,"githubUrl":"https://github.com/crowdsecurity/crowdsec/blob/909b5157986a2b2c2163300fdaef5ed01289f7d2/pkg/acquisition/modules/kinesis/run.go#L270-L306","documentation":"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.","triggerScenarios":"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).","commonSituations":"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.","solutions":["Check IAM permissions for kinesis:DeregisterStreamConsumer on the consumer/stream.","If the consumer no longer exists, make DeregisterConsumer tolerate ResourceNotFoundException and treat it as success.","Verify region/account consistency of the persisted consumer ARN.","Retry on throttling with backoff; then rerun the datasource."],"exampleFix":"// before: any error aborts startup\nerr = s.DeregisterConsumer(ctx)\n// after: tolerate already-gone consumers\nvar rnfe *kinTypes.ResourceNotFoundException\nif err != nil && !errors.As(err, &rnfe) {\n\treturn fmt.Errorf(\"cannot deregister consumer: %w\", err)\n}","handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"var rnfe *kinTypes.ResourceNotFoundException\nif err := deregister(ctx); err != nil && !errors.As(err, &rnfe) {\n\treturn fmt.Errorf(\"cannot deregister consumer: %w\", err)\n}","preventionTips":["Treat ResourceNotFound on deregister as success.","Grant kinesis:DeregisterStreamConsumer in IAM.","Avoid deleting the consumer manually between runs; let the datasource manage it.","Use unique consumer names to prevent cross-instance races."],"tags":["aws","kinesis","iam","lifecycle"],"backgroundTag":"permission-denied","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"}