{"record":{"id":"4ba8e6270dd5fbc1","repo":"apache/beam","slug":"executor-service-is-taking-long-time-to-shutdown-will-retry","errorCode":null,"errorMessage":"Executor service is taking long time to shutdown, will retry. {} attempts left","messagePattern":"Executor service is taking long time to shutdown, will retry\\. (.+?) attempts left","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":243,"sourceCode":"          ATTEMPTS_TO_SHUTDOWN);\n      executorService.shutdownNow();\n      awaitTermination();\n    }\n  }\n\n  private void awaitTermination() {\n    int attemptsLeft = ATTEMPTS_TO_SHUTDOWN;\n    boolean isTerminated = executorService.isTerminated();\n\n    while (!isTerminated && attemptsLeft-- > 0) {\n      try {\n        isTerminated = executorService.awaitTermination(10, TimeUnit.SECONDS);\n      } catch (InterruptedException e) {\n        LOG.error(\"Interrupted while waiting for the executor service to shutdown\");\n        throw new RuntimeException(e);\n      }\n      if (!isTerminated && attemptsLeft > 0) {\n        LOG.warn(\n            \"Executor service is taking long time to shutdown, will retry. {} attempts left\",\n            attemptsLeft);\n      }\n    }\n  }\n\n  Instant getWatermark() {\n    return getMinTimestamp(ShardRecordsIterator::getShardWatermark);\n  }\n\n  Instant getLatestRecordTimestamp() {\n    return getMinTimestamp(ShardRecordsIterator::getLatestRecordTimestamp);\n  }\n\n  private Instant getMinTimestamp(Function<ShardRecordsIterator, Instant> timestampExtractor) {\n    return minTimestamp(shardIteratorsMap.get().values().stream().map(timestampExtractor));\n  }\n","sourceCodeStart":225,"sourceCodeEnd":261,"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#L225-L261","documentation":"awaitTermination repeatedly waits 10-second intervals for the executor to finish. When a wait expires but attempts remain, it logs this warning and retries; shutdown proceeds once attempts are exhausted (escalating to shutdownNow in stop()).","triggerScenarios":"Executor threads running shard reading loops don't terminate within 10 seconds per attempt during ShardReadersPool.stop(), typically because they are blocked on a full records queue or an in-flight Kinesis request.","commonSituations":"Slow consumer pipelines during teardown; long-running getRecords calls; Kinesis endpoint latency; threads that swallow interrupts.","solutions":["Wait it out — shutdown escalates to shutdownNow after remaining attempts elapse","Keep downstream consumption healthy so the reader queue drains and tasks exit promptly","Set aggressive client-level request timeouts so in-flight getRecords calls cannot stall termination","Check for application code holding interrupts (catching InterruptedException without re-interrupting) inside custom DoFns"],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"try {\n  executorService.shutdown();\n  if (!executorService.awaitTermination(60, TimeUnit.SECONDS)) {\n    executorService.shutdownNow();\n  }\n} catch (InterruptedException e) {\n  Thread.currentThread().interrupt();\n  executorService.shutdownNow();\n}","preventionTips":["Same as 6134: drain queues, bound request timeouts, respect interrupts","Avoid sharing one executor across many slow sinks","Check SQS/Kinesis endpoint latency before scheduled shutdown windows"],"tags":["kinesis","thread-pool","shutdown","timeout"],"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"}