apache/beam · info · KinesisShardClosedException

Shard iterator reached end of the shard: streamName=

Error message

Shard iterator reached end of the shard: streamName=%s, shardId=%s

What it means

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.

Solutions

1. Catch KinesisShardClosedException and treat the shard as finished; stop reading it and move to other shards.
2. In Beam Kinesis IO this is handled automatically — ensure you use the provided KinesisReader rather than calling readNextBatch directly.
3. If you expect more data, verify via listShards whether the shard was merged/split and start reading the successor shards.

Example fix

try {
  List<KinesisRecord> batch = iterator.readNextBatch();
} catch (KinesisShardClosedException e) {
  // shard is done; mark complete and continue with other shards
}

try {
  List<KinesisRecord> batch = iterator.readNextBatch();
} catch (KinesisShardClosedException e) {
  // shard exhausted; stop consuming this shard
}

When it happens

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

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

Understand the failure class

Background: "Invalid state transition" errors: "status must be X, actually Y", "already rejected/charging/uninstalled", "cannot ... while running" — what they mean when a library rejects your call — this error's family across 31 libraries.

Related errors

• Executor service is taking long time to shutdown, will…
• Executor service was not completely terminated after
• Expected input schema with a 'data' (BYTES) field, a…
• Failed to refresh shards.
• Interrupted while waiting for KinesisRecord from the buffer

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/278c6f42e34d4928. Report an issue: GitHub.

Appendix: source

Thrown at sdks/java/io/amazon-web-services2/src/main/java/org/apache/beam/sdk/io/aws2/kinesis/ShardRecordsIterator.java:81

      ShardCheckpoint initialCheckpoint,
      SimplifiedKinesisClient simplifiedKinesisClient,
      WatermarkPolicyFactory watermarkPolicyFactory,
      RecordFilter filter)
      throws TransientKinesisException {
    this.checkpoint = new AtomicReference<>(checkNotNull(initialCheckpoint, "initialCheckpoint"));
    this.filter = checkNotNull(filter, "filter");
    this.kinesis = checkNotNull(simplifiedKinesisClient, "simplifiedKinesisClient");
    this.streamName = initialCheckpoint.getStreamName();
    this.shardId = initialCheckpoint.getShardId();
    this.shardIterator = initialCheckpoint.getShardIterator(kinesis);
    this.watermarkPolicy = watermarkPolicyFactory.createWatermarkPolicy();
    this.watermarkPolicyFactory = watermarkPolicyFactory;
  }

  List<KinesisRecord> readNextBatch()
      throws TransientKinesisException, KinesisShardClosedException {
    if (shardIterator == null) {
      throw new KinesisShardClosedException(
          String.format(
              "Shard iterator reached end of the shard: streamName=%s, shardId=%s",
              streamName, shardId));
    }
    GetKinesisRecordsResult response = fetchRecords();
    LOG.debug(
        "Fetched {} new records from shard: streamName={}, shardId={}",
        response.getRecords().size(),
        streamName,
        shardId);

    List<KinesisRecord> filteredRecords = filter.apply(response.getRecords(), checkpoint.get());
    return filteredRecords;
  }

  private GetKinesisRecordsResult fetchRecords() throws TransientKinesisException {
    try {
      GetKinesisRecordsResult response = kinesis.getRecords(shardIterator, streamName, shardId);

View on GitHub (pinned to 12126d8942)