{"record":{"id":"2126351d2dfd9fbd","repo":"crowdsecurity/crowdsec","slug":"cannot-close-kinesis-subscribed-shard-reader-w","errorCode":null,"errorMessage":"cannot close kinesis subscribed shard reader: %w","messagePattern":"cannot close kinesis subscribed shard reader: %w","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"warning","filePath":"pkg/acquisition/modules/kinesis/run.go","lineNumber":225,"sourceCode":"\t\t\tout <- evt\n\t\t}\n\t}\n}\n\nfunc (s *Source) ReadFromSubscription(reader kinesis.SubscribeToShardEventStreamReader, out chan pipeline.Event, shardID string, streamName string) error {\n\tlogger := s.logger.WithField(\"shard_id\", shardID)\n\t// ghetto sync, kinesis allows to subscribe to a closed shard, which will make the goroutine exit immediately\n\t// and we won't be able to start a new one if this is the first one started by the tomb\n\t// TODO: look into parent shards to see if a shard is closed before starting to read it ?\n\ttime.Sleep(time.Second)\n\n\tfor {\n\t\tselect {\n\t\tcase <-s.shardReaderTomb.Dying():\n\t\t\tlogger.Infof(\"Subscribed shard reader is dying\")\n\n\t\t\tif err := reader.Close(); err != nil {\n\t\t\t\treturn fmt.Errorf(\"cannot close kinesis subscribed shard reader: %w\", err)\n\t\t\t}\n\n\t\t\treturn nil\n\t\tcase event, ok := <-reader.Events():\n\t\t\tif !ok {\n\t\t\t\tlogger.Infof(\"Event chan has been closed\")\n\t\t\t\treturn nil\n\t\t\t}\n\n\t\t\tswitch et := event.(type) {\n\t\t\tcase *kinTypes.SubscribeToShardEventStreamMemberSubscribeToShardEvent:\n\t\t\t\ts.ParseAndPushRecords(et.Value.Records, out, logger, shardID)\n\t\t\tdefault:\n\t\t\t\tlogger.Infof(\"unhandled SubscribeToShard event: %T\", et)\n\t\t\t}\n\t\t}\n\t}\n}","sourceCodeStart":207,"sourceCodeEnd":243,"githubUrl":"https://github.com/crowdsecurity/crowdsec/blob/909b5157986a2b2c2163300fdaef5ed01289f7d2/pkg/acquisition/modules/kinesis/run.go#L207-L243","documentation":"Returned by ReadFromSubscription when the shard reader (a Kinesis Client Library-style subscriber reader) fails to Close after the reader's tombstone dies, i.e. during graceful shutdown of the subscribed shard reader. A Close failure means resources (connections, leases) may be left dangling.","triggerScenarios":"Source shutdown (ctx cancelled / tomb killed) triggers reader.Close(), which returns an error — typically from failing to flush/checkpoint or cleanly release the underlying Kinesis subscription connection.","commonSituations":"Shutting down while a GetRecords/SubscribeToShard call is in flight; network already severed so the close handshake fails; the reader was already closed elsewhere (double-close).","solutions":["Check the wrapped cause: if it's a network error during shutdown, it is usually benign since the process is exiting — verify records were checkpointed before shutdown.","Ensure the shutdown path doesn't close the reader twice (e.g. from both the tomb callback and the read loop).","Improve network stability or increase client timeouts so in-flight calls finish before close.","If it happens on AWS side consistently, check CloudWatch metrics for SubscribeToShard flapping and the consumer's lease health."],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"case <-s.shardReaderTomb.Dying():\n    if err := reader.Close(); err != nil {\n        // shutdown path: log and continue, don't mask a successful run\n        logger.WithError(err).Warn(\"kinesis shard reader close failed during shutdown\")\n    }\n    return nil","preventionTips":["Checkpoint records before killing the tomb so close has nothing pending to flush.","Avoid closing the reader from two code paths (double-close).","Increase client timeouts so in-flight SubscribeToShard reads finish during shutdown.","Treat close failures during process exit as warnings, not fatal, unless checkpointing is affected."],"tags":["aws","kinesis","shutdown"],"backgroundTag":"broken-pipe","analyzedSha":"909b5157986a2b2c2163300fdaef5ed01289f7d2","analyzedAt":"2026-09-06T12:27:26.012Z","contentChangedAt":"2026-09-06T12:27:26.012Z","schemaVersion":2},"datasetVersion":"2026-09-14T00:17:10.932Z"}