{"record":{"id":"278c6f42e34d4928","repo":"apache/beam","slug":"shard-iterator-reached-end-of-the-shard-streamname-s-shardid","errorCode":null,"errorMessage":"Shard iterator reached end of the shard: streamName=%s, shardId=%s","messagePattern":"Shard iterator reached end of the shard: streamName=(.+?), shardId=(.+?)","errorType":"exception","errorClass":"KinesisShardClosedException","httpStatus":null,"severity":"info","filePath":"sdks/java/io/amazon-web-services2/src/main/java/org/apache/beam/sdk/io/aws2/kinesis/ShardRecordsIterator.java","lineNumber":81,"sourceCode":"      ShardCheckpoint initialCheckpoint,\n      SimplifiedKinesisClient simplifiedKinesisClient,\n      WatermarkPolicyFactory watermarkPolicyFactory,\n      RecordFilter filter)\n      throws TransientKinesisException {\n    this.checkpoint = new AtomicReference<>(checkNotNull(initialCheckpoint, \"initialCheckpoint\"));\n    this.filter = checkNotNull(filter, \"filter\");\n    this.kinesis = checkNotNull(simplifiedKinesisClient, \"simplifiedKinesisClient\");\n    this.streamName = initialCheckpoint.getStreamName();\n    this.shardId = initialCheckpoint.getShardId();\n    this.shardIterator = initialCheckpoint.getShardIterator(kinesis);\n    this.watermarkPolicy = watermarkPolicyFactory.createWatermarkPolicy();\n    this.watermarkPolicyFactory = watermarkPolicyFactory;\n  }\n\n  List<KinesisRecord> readNextBatch()\n      throws TransientKinesisException, KinesisShardClosedException {\n    if (shardIterator == null) {\n      throw new KinesisShardClosedException(\n          String.format(\n              \"Shard iterator reached end of the shard: streamName=%s, shardId=%s\",\n              streamName, shardId));\n    }\n    GetKinesisRecordsResult response = fetchRecords();\n    LOG.debug(\n        \"Fetched {} new records from shard: streamName={}, shardId={}\",\n        response.getRecords().size(),\n        streamName,\n        shardId);\n\n    List<KinesisRecord> filteredRecords = filter.apply(response.getRecords(), checkpoint.get());\n    return filteredRecords;\n  }\n\n  private GetKinesisRecordsResult fetchRecords() throws TransientKinesisException {\n    try {\n      GetKinesisRecordsResult response = kinesis.getRecords(shardIterator, streamName, shardId);","sourceCodeStart":63,"sourceCodeEnd":99,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/amazon-web-services2/src/main/java/org/apache/beam/sdk/io/aws2/kinesis/ShardRecordsIterator.java#L63-L99","documentation":"readNextBatch in ShardRecordsIterator throws KinesisShardClosedException when the cached shardIterator is null, meaning a previous getRecords call returned a null NextShardIterator — the shard has been closed (or records were consumed past the last sequence number) and no more data will ever be available from it. It is a normal end-of-shard signal, not a transport failure.","triggerScenarios":"Calling readNextBatch after an earlier fetch response had NextShardIterator == null (shard closed due to resharding/merge/split, or all records up to the sequence number were consumed, e.g. reading with a very recent AT_TIMESTAMP or LATEST iterator after shard end).","commonSituations":"Kinesis stream resharding closed a shard while the Beam consumer was still reading it; consumers attempting to read a shard beyond its last record; custom integrations looping readNextBatch without checking shard-closed state.","solutions":["Catch KinesisShardClosedException and treat the shard as finished; stop reading it and move to other shards.","In Beam Kinesis IO this is handled automatically — ensure you use the provided KinesisReader rather than calling readNextBatch directly.","If you expect more data, verify via listShards whether the shard was merged/split and start reading the successor shards."],"exampleFix":"try {\n  List<KinesisRecord> batch = iterator.readNextBatch();\n} catch (KinesisShardClosedException e) {\n  // shard is done; mark complete and continue with other shards\n}","handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"try {\n  List<KinesisRecord> batch = iterator.readNextBatch();\n} catch (KinesisShardClosedException e) {\n  // shard exhausted; stop consuming this shard\n}","preventionTips":["Treat shard-closed as a normal terminal state in consumption loops","Track resharding via listShards and switch to successor shards"],"tags":["kinesis","shard-closed","stream-resharding"],"backgroundTag":"invalid-state-transition","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}