crowdsecurity/crowdsec · warning

cannot close kinesis subscribed shard reader: %w

Error message

cannot close kinesis subscribed shard reader: %w

What it means

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.

Source

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

			out <- evt
		}
	}
}

func (s *Source) ReadFromSubscription(reader kinesis.SubscribeToShardEventStreamReader, out chan pipeline.Event, shardID string, streamName string) error {
	logger := s.logger.WithField("shard_id", shardID)
	// ghetto sync, kinesis allows to subscribe to a closed shard, which will make the goroutine exit immediately
	// and we won't be able to start a new one if this is the first one started by the tomb
	// TODO: look into parent shards to see if a shard is closed before starting to read it ?
	time.Sleep(time.Second)

	for {
		select {
		case <-s.shardReaderTomb.Dying():
			logger.Infof("Subscribed shard reader is dying")

			if err := reader.Close(); err != nil {
				return fmt.Errorf("cannot close kinesis subscribed shard reader: %w", err)
			}

			return nil
		case event, ok := <-reader.Events():
			if !ok {
				logger.Infof("Event chan has been closed")
				return nil
			}

			switch et := event.(type) {
			case *kinTypes.SubscribeToShardEventStreamMemberSubscribeToShardEvent:
				s.ParseAndPushRecords(et.Value.Records, out, logger, shardID)
			default:
				logger.Infof("unhandled SubscribeToShard event: %T", et)
			}
		}
	}
}

View on GitHub (pinned to 909b515798)

Solutions

1. 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.
2. Ensure the shutdown path doesn't close the reader twice (e.g. from both the tomb callback and the read loop).
3. Improve network stability or increase client timeouts so in-flight calls finish before close.
4. If it happens on AWS side consistently, check CloudWatch metrics for SubscribeToShard flapping and the consumer's lease health.

case <-s.shardReaderTomb.Dying():
    if err := reader.Close(); err != nil {
        // shutdown path: log and continue, don't mask a successful run
        logger.WithError(err).Warn("kinesis shard reader close failed during shutdown")
    }
    return nil

When it happens

Trigger: 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.

Common situations: 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).

Related errors

• stream_name is mandatory when use_enhanced_fanout is false
• stream_arn is mandatory when use_enhanced_fanout is true
• consumer_name is mandatory when use_enhanced_fanout is true
• stream_arn and stream_name are mutually exclusive
• cannot create kinesis client: %w

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