Interrupted while waiting for KinesisRecord from the buffer

Error message

Interrupted while waiting for KinesisRecord from the buffer

What it means

ShardReadersPool.nextRecord waits on a buffered queue for the next KinesisRecord. On InterruptedException, the wait is abandoned, a warning logged (without the stack trace), and CustomOptional.absent() is returned to signal no record available. The caller typically treats absent as end-of-shard/no-data for this poll; repeated occurrences during shutdown are expected.

Source

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

      }
    }
    LOG.info("Kinesis Shard read loop has finished");
  }

  CustomOptional<KinesisRecord> nextRecord() {
    try {
      KinesisRecord record = recordsQueue.poll(QUEUE_POLL_TIMEOUT_MS, MILLISECONDS);
      if (record == null) {
        return CustomOptional.absent();
      }
      shardIteratorsMap.get().get(record.getShardId()).ackRecord(record);

      // numberOfRecordsInAQueueByShard contains the counter for a given shard until the shard is
      // closed and then it's counter reaches 0. Thus the access here is safe
      numberOfRecordsInAQueueByShard.get(record.getShardId()).decrementAndGet();
      return CustomOptional.of(record);
    } catch (InterruptedException e) {
      LOG.warn("Interrupted while waiting for KinesisRecord from the buffer");
      return CustomOptional.absent();
    }
  }

  void stop() {
    LOG.info("Closing shard iterators pool");
    poolOpened.set(false);
    executorService.shutdown();
    awaitTermination();
    if (!executorService.isTerminated()) {
      LOG.warn(
          "Executor service was not completely terminated after {} attempts, trying to forcibly stop it.",
          ATTEMPTS_TO_SHUTDOWN);
      executorService.shutdownNow();
      awaitTermination();
    }
  }

View on GitHub (pinned to 12126d8942)

Solutions

No action if it occurs during shutdown — absent() is the designed response.
If it occurs mid-run unexpectedly, find the interrupting component (lifecycle logs).
Check whether the buffer timeout is too short relative to shard data rates so waits are longer than shutdown windows.
Resume reading; sequence numbers persist so no records are lost.

Example fix

// before
} catch (InterruptedException e) {
  LOG.warn("Interrupted while waiting for KinesisRecord from the buffer");
  return CustomOptional.absent();
}
// after
} catch (InterruptedException e) {
  LOG.warn("Interrupted while waiting for KinesisRecord from the buffer");
  Thread.currentThread().interrupt(); // restore interrupt flag
  return CustomOptional.absent();
}

Defensive patterns

Strategy: try-catch

Try / catch

CustomOptional<KinesisRecord> rec = pool.nextRecord(shardId);
if (!rec.isPresent()) {
  // absent may mean interrupted wait; check shard status / retry
}

Prevention

Treat absent() as 'no record now', not necessarily end-of-shard
Restore the interrupt flag in custom queue consumers
Ensure shutdown ordering: stop producers, then consumers
Size queue wait times relative to expected shard data rates

When it happens

Trigger: nextRecord blocks on the record queue's poll/take and the thread is interrupted — usually by stop() closing the pool or runner teardown while no record was available within the wait window.

Common situations: Unbounded reader shutdown mid-poll; drain/checkpoint operations interrupting the consumer thread; idle shards with no records when the thread gets interrupted.

Related errors

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