{"record":{"id":"1c8877294bc07b5f","repo":"apache/hadoop","slug":"stream-closed-or-unbuffer-is-called","errorCode":null,"errorMessage":"Stream closed or unbuffer is called","messagePattern":"Stream closed or unbuffer is called","errorType":"exception","errorClass":"InterruptedIOException","httpStatus":null,"severity":"warning","filePath":"hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AInputStream.java","lineNumber":1197,"sourceCode":"      throw ex;\n    } finally {\n      tracker.close();\n    }\n    changeTracker.processResponse(objectRange.response(), operationName,\n            position);\n    return objectRange;\n  }\n\n  /**\n   * Check if vectored io operation has been stooped. This happens\n   * when the stream is closed or unbuffer is called.\n   * @throws InterruptedIOException throw InterruptedIOException such\n   *                                that all running vectored io is\n   *                                terminated thus releasing resources.\n   */\n  private void checkIfVectoredIOStopped() throws InterruptedIOException {\n    if (stopVectoredIOOperations.get()) {\n      throw new InterruptedIOException(\"Stream closed or unbuffer is called\");\n    }\n  }\n\n  @Override\n  public synchronized void setReadahead(Long readahead) {\n    this.readahead = validateReadahead(readahead);\n  }\n\n  /**\n   * Get the current readahead value.\n   * @return a non-negative readahead value\n   */\n  public synchronized long getReadahead() {\n    return readahead;\n  }\n\n  /**\n   * Calculate the limit for a get request, based on input policy","sourceCodeStart":1179,"sourceCodeEnd":1215,"githubUrl":"https://github.com/apache/hadoop/blob/2add9630210752f88ceb1bb74eb65e37bf41da8e/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AInputStream.java#L1179-L1215","documentation":"InterruptedIOException thrown by S3AInputStream.checkIfVectoredIOStopped(), which is polled inside vectored-read loops. Setting the stopVectoredIOOperations flag - done by close() and unbuffer() - makes all in-flight vectored reads terminate promptly so HTTP streams and buffers are released instead of leaking.","triggerScenarios":"Another thread calls close() or unbuffer() on the stream while readVectored() ranges are still being read; the engine releases resources between query stages while a vectored read is in flight; task cancellation closing inputs mid-read.","commonSituations":"Hive/Spark calling unbuffer() after a query phase or on task switch; shared streams where one consumer closes while another still reads; preemption cancelling tasks that own the stream.","solutions":["Treat it as an interruption, not data corruption: reopen the stream (fs.open) and reissue the readVectored when the data is still needed","Fix stream ownership so the reader finishes or cancels its vectored-read futures before anyone calls unbuffer()/close()","If you call unbuffer() deliberately, consume or cancel the outstanding futures first","Propagate the interrupt properly in worker threads so thread pools are not left in a bad state"],"exampleFix":"// before\nList<CompletableFuture<ByteBuffer>> futs = in.readVectored(ranges, ByteBuffer::allocate);\nengine.unbuffer(in);            // or close() from another thread\nfuts.get(0).join();            // InterruptedIOException: Stream closed or unbuffer is called\n\n// after - consume futures before releasing the stream\nList<CompletableFuture<ByteBuffer>> futs = in.readVectored(ranges, ByteBuffer::allocate);\nByteBuffer b = futs.get(0).join();\nin.unbuffer();                  // release resources after consumption","handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"try {\n  in.readVectored(ranges, ByteBuffer::allocate).forEach(f -> f.join());\n} catch (CompletionException e) {\n  if (e.getCause() instanceof InterruptedIOException) {\n    in = fs.open(path); // stream was closed/unbuffered: reopen if data is still needed\n  } else {\n    throw e;\n  }\n}","preventionTips":["Consume or cancel vectored-read futures before calling unbuffer() or close()","Give each concurrent consumer its own stream","Handle InterruptedIOException as 'resource released', never as data corruption"],"tags":["s3a","hadoop-aws","vectored-read","interruptedioexception","unbuffer","stream-lifecycle"],"backgroundTag":"io-operation-interrupted","analyzedSha":"2add9630210752f88ceb1bb74eb65e37bf41da8e","analyzedAt":"2026-08-22T19:55:07.957Z","schemaVersion":2},"datasetVersion":"2026-08-22T20:17:22.307Z"}