{"record":{"id":"ee7b70c452565360","repo":"apache/hadoop","slug":"this-stream-is-already-closed","errorCode":null,"errorMessage":"this stream is already closed","messagePattern":"this stream is already closed","errorType":"exception","errorClass":"IOException","httpStatus":null,"severity":"error","filePath":"hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/impl/prefetch/CachingBlockManager.java","lineNumber":156,"sourceCode":"\n  /**\n   * Gets the block having the given {@code blockNumber}.\n   *\n   * @throws IllegalArgumentException if blockNumber is negative.\n   */\n  @Override\n  public BufferData get(int blockNumber) throws IOException {\n    checkNotNegative(blockNumber, \"blockNumber\");\n\n    BufferData data;\n    final int maxRetryDelayMs = bufferPoolSize * 120 * 1000;\n    final int statusUpdateDelayMs = 120 * 1000;\n    Retryer retryer = new Retryer(10, maxRetryDelayMs, statusUpdateDelayMs);\n    boolean done;\n\n    do {\n      if (closed) {\n        throw new IOException(\"this stream is already closed\");\n      }\n\n      data = bufferPool.acquire(blockNumber);\n      done = getInternal(data);\n\n      if (retryer.updateStatus()) {\n        LOG.warn(\"waiting to get block: {}\", blockNumber);\n        LOG.info(\"state = {}\", this.toString());\n      }\n    }\n    while (!done && retryer.continueRetry());\n\n    if (done) {\n      return data;\n    } else {\n      String message = String.format(\"Wait failed for get(%d)\", blockNumber);\n      throw new IllegalStateException(message);\n    }","sourceCodeStart":138,"sourceCodeEnd":174,"githubUrl":"https://github.com/apache/hadoop/blob/2add9630210752f88ceb1bb74eb65e37bf41da8e/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/impl/prefetch/CachingBlockManager.java#L138-L174","documentation":"CachingBlockManager.get(blockNumber) re-checks the manager's closed flag on every iteration of its retry loop; if the owning prefetch stream was closed while get() is still running (typically from another thread), it aborts with IOException(\"this stream is already closed\") rather than returning data.","triggerScenarios":"One thread closes the prefetch input stream while another is blocked in read()/get() waiting for a block; task cancellation closing inputs mid-read; watchdog code closing \"stuck\" streams that are merely waiting on a slow block fetch.","commonSituations":"Sharing one stream across threads without a close handshake; MR/Spark task-kill paths closing inputs while reader threads drain; timeout-driven cleanup racing legitimate slow reads.","solutions":["Give the stream a single owner: only the owner closes, and only after readers finish","In cancellation paths, signal readers to stop and join them before closing","Treat this IOException in reader threads as a cancellation signal, not a data error","Open a separate stream per thread instead of sharing one"],"exampleFix":"// before\nnew Thread(() -> { try { in.read(); } catch (IOException e) { /* ... */ } }).start();\nin.close();            // racing reader -> \"this stream is already closed\"\n\n// after\nThread reader = new Thread(() -> { try { in.read(); } catch (IOException e) { /* ... */ } });\nreader.start();\nreader.join();         // reader finished before close\nin.close();","handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"in reader threads, catch IOException with message \"this stream is already closed\" and interpret it as cancellation/shutdown — exit the reader cleanly instead of reporting a data error.","preventionTips":["Single owner closes the stream, and only after readers finish (join them)","In cancellation paths, signal readers before closing inputs","Open one stream per thread rather than sharing"],"tags":["hadoop","prefetch","stream-lifecycle","concurrency"],"backgroundTag":"stream-closed","analyzedSha":"2add9630210752f88ceb1bb74eb65e37bf41da8e","analyzedAt":"2026-08-22T19:55:07.957Z","schemaVersion":2},"datasetVersion":"2026-08-22T20:17:22.307Z"}