{"record":{"id":"9c078806cc591d1c","repo":"apache/beam","slug":"thread-was-interrupted-finishing-the-read-loop","errorCode":null,"errorMessage":"Thread was interrupted, finishing the read loop","messagePattern":"Thread was interrupted, finishing the read loop","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":182,"sourceCode":"        } catch (KinesisShardClosedException e) {\n          LOG.info(\n              \"Shard iterator for {} shard is closed, finishing the read loop\",\n              shardRecordsIterator.getShardId(),\n              e);\n          // Wait until all records from already closed shard are taken from the buffer and only\n          // then start reading successive shards. This guarantees that checkpoints will contain\n          // either parent or child shard and never both. Such approach allows for more\n          // straightforward checkpoint restoration than in a case when new shards are read\n          // immediately.\n          waitUntilAllShardRecordsRead(shardRecordsIterator);\n          readFromSuccessiveShards(shardRecordsIterator);\n          break;\n        }\n      } catch (KinesisClientThrottledException e) {\n        try {\n          rateLimiter.onThrottle(e);\n        } catch (InterruptedException ex) {\n          LOG.warn(\"Thread was interrupted, finishing the read loop\", ex);\n          Thread.currentThread().interrupt();\n          break;\n        }\n      } catch (TransientKinesisException e) {\n        LOG.warn(\"Transient exception occurred.\", e);\n      } catch (InterruptedException e) {\n        LOG.warn(\"Thread was interrupted, finishing the read loop\", e);\n        Thread.currentThread().interrupt();\n        break;\n      } catch (Throwable e) {\n        LOG.error(\"Unexpected exception occurred\", e);\n      }\n    }\n    LOG.info(\"Kinesis Shard read loop has finished\");\n  }\n\n  CustomOptional<KinesisRecord> nextRecord() {\n    try {","sourceCodeStart":164,"sourceCodeEnd":200,"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#L164-L200","documentation":"In ShardReadersPool.readLoop, when a KinesisClientThrottledException occurs the rate limiter's onThrottle() sleeps; if that sleep is interrupted, the thread's interrupt status is restored and the read loop is terminated gracefully. This is the shutdown path, not a failure — it usually indicates the reader thread was asked to stop while throttled.","triggerScenarios":"readLoop gets throttled by Kinesis, calls rateLimiter.onThrottle(e) which blocks, and the thread is interrupted (pool stop()/shutdown or runner teardown) during that wait.","commonSituations":"Pipeline stoppage/drain while Kinesis is throttling the consumer; aggressive record limiter configuration causing long throttle sleeps that overlap shutdown; too many consumers sharing a shard.","solutions":["No fix needed if this happens at shutdown — the loop exits cleanly.","If it happens unexpectedly, check which component interrupts the reader thread (runner checkpoint/cancel).","Reduce throttling: lower consumer parallelism or increase the rate limiter's capacity/burst settings.","Retry the pipeline; throttling is transient."],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"try {\n  rateLimiter.onThrottle(e);\n} catch (InterruptedException ex) {\n  Thread.currentThread().interrupt(); // always restore flag before exiting\n  break;\n}","preventionTips":["Configure rate limiters so throttle sleeps are short relative to shutdown time","Reduce shard contention to avoid KinesisClientThrottledException","Coordinate stop() with read loop lifecycle to avoid interrupts mid-throttle","Scale shards/consumers to stay under Kinesis quotas"],"tags":["aws","kinesis","interruption","shutdown"],"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"}