{"record":{"id":"bf6f80b32a6bf99a","repo":"apache/beam","slug":"interrupted-while-waiting-for-kinesisrecord-from-the-buffer","errorCode":null,"errorMessage":"Interrupted while waiting for KinesisRecord from the buffer","messagePattern":"Interrupted while waiting for KinesisRecord from the buffer","errorType":"console","errorClass":null,"httpStatus":null,"severity":"warning","filePath":"sdks/java/io/amazon-web-services2/src/main/java/org/apache/beam/sdk/io/aws2/kinesis/ShardReadersPool.java","lineNumber":212,"sourceCode":"      }\n    }\n    LOG.info(\"Kinesis Shard read loop has finished\");\n  }\n\n  CustomOptional<KinesisRecord> nextRecord() {\n    try {\n      KinesisRecord record = recordsQueue.poll(QUEUE_POLL_TIMEOUT_MS, MILLISECONDS);\n      if (record == null) {\n        return CustomOptional.absent();\n      }\n      shardIteratorsMap.get().get(record.getShardId()).ackRecord(record);\n\n      // numberOfRecordsInAQueueByShard contains the counter for a given shard until the shard is\n      // closed and then it's counter reaches 0. Thus the access here is safe\n      numberOfRecordsInAQueueByShard.get(record.getShardId()).decrementAndGet();\n      return CustomOptional.of(record);\n    } catch (InterruptedException e) {\n      LOG.warn(\"Interrupted while waiting for KinesisRecord from the buffer\");\n      return CustomOptional.absent();\n    }\n  }\n\n  void stop() {\n    LOG.info(\"Closing shard iterators pool\");\n    poolOpened.set(false);\n    executorService.shutdown();\n    awaitTermination();\n    if (!executorService.isTerminated()) {\n      LOG.warn(\n          \"Executor service was not completely terminated after {} attempts, trying to forcibly stop it.\",\n          ATTEMPTS_TO_SHUTDOWN);\n      executorService.shutdownNow();\n      awaitTermination();\n    }\n  }\n","sourceCodeStart":194,"sourceCodeEnd":230,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/amazon-web-services2/src/main/java/org/apache/beam/sdk/io/aws2/kinesis/ShardReadersPool.java#L194-L230","documentation":"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.","triggerScenarios":"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.","commonSituations":"Unbounded reader shutdown mid-poll; drain/checkpoint operations interrupting the consumer thread; idle shards with no records when the thread gets interrupted.","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."],"exampleFix":"// before\n} catch (InterruptedException e) {\n  LOG.warn(\"Interrupted while waiting for KinesisRecord from the buffer\");\n  return CustomOptional.absent();\n}\n// after\n} catch (InterruptedException e) {\n  LOG.warn(\"Interrupted while waiting for KinesisRecord from the buffer\");\n  Thread.currentThread().interrupt(); // restore interrupt flag\n  return CustomOptional.absent();\n}","handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"CustomOptional<KinesisRecord> rec = pool.nextRecord(shardId);\nif (!rec.isPresent()) {\n  // absent may mean interrupted wait; check shard status / retry\n}","preventionTips":["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"],"tags":["aws","kinesis","interruption","queue"],"backgroundTag":"thread-interrupted","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T21:17:11.552Z"}