{"record":{"id":"152609cd6ebc0dea","repo":"apache/druid","slug":"thread-interrupted-while-adding-to-queue","errorCode":null,"errorMessage":"Thread interrupted while adding to queue","messagePattern":"Thread interrupted while adding to queue","errorType":"console","errorClass":null,"httpStatus":null,"severity":"warning","filePath":"processing/src/main/java/org/apache/druid/java/util/http/client/response/SequenceInputStreamResponseHandler.java","lineNumber":121,"sourceCode":"  public ClientResponse<InputStream> handleChunk(\n      ClientResponse<InputStream> clientResponse,\n      HttpChunk chunk,\n      long chunkNum\n  )\n  {\n    final ChannelBuffer channelBuffer = chunk.getContent();\n    final int bytes = channelBuffer.readableBytes();\n    if (bytes > 0) {\n      try (ChannelBufferInputStream channelStream = new ChannelBufferInputStream(channelBuffer)) {\n        queue.put(channelStream);\n        // Queue.size() can be expensive in some implementations, but LinkedBlockingQueue.size is just an AtomicLong\n        log.debug(\"Added stream. Queue length %d\", queue.size());\n      }\n      catch (IOException e) {\n        throw new RuntimeException(e);\n      }\n      catch (InterruptedException e) {\n        log.warn(e, \"Thread interrupted while adding to queue\");\n        Thread.currentThread().interrupt();\n        throw new RuntimeException(e);\n      }\n      byteCount.addAndGet(bytes);\n    } else {\n      log.debug(\"Skipping zero length chunk\");\n    }\n    return clientResponse;\n  }\n\n  @Override\n  public ClientResponse<InputStream> done(ClientResponse<InputStream> clientResponse)\n  {\n    synchronized (done) {\n      try {\n        // An empty byte array is put at the end to give the SequenceInputStream.close() as something to close out\n        // after done is set to true, regardless of the rest of the stream's state.\n        queue.put(ByteSource.empty().openStream());","sourceCodeStart":103,"sourceCodeEnd":139,"githubUrl":"https://github.com/apache/druid/blob/9b90983fd291f26935af934383ce360473179e4d/processing/src/main/java/org/apache/druid/java/util/http/client/response/SequenceInputStreamResponseHandler.java#L103-L139","documentation":"In SequenceInputStreamResponseHandler.handleChunk, each downloaded chunk is opened as a stream and put into the queue with queue.put(), which blocks when the queue is full. If the producer thread is interrupted while blocked on put(), the code logs a warning, re-interrupts, and wraps the InterruptedException in a RuntimeException. This signals the response processing was cancelled mid-chunk.","triggerScenarios":"The thread running handleChunk is interrupted while blocked on queue.put() because the consumer stopped reading (queue full) and someone interrupted the producer, e.g. request cancellation or client disconnect detected mid-response.","commonSituations":"Downstream consumer crashed/closed while the HTTP client continued downloading, filling the bounded queue; query cancellation; container/executor shutdown during a long streaming download.","solutions":["Determine who interrupted the producer thread; usually a legitimate cancellation (client disconnect, query cancel)","Ensure consumers read or close the SequenceInputStream promptly to keep the queue draining","Avoid Future.cancel(true)/Thread.interrupt() on the HTTP response-processing thread except for real cancellations","Check for timeouts configured shorter than the download duration that trigger interrupts"],"exampleFix":"// before: abandoning the stream mid-download\nhttpClient.go(request, handler).get(); // no timeout/cancel management\n// after: register cancellation and close on timeout\nListenableFuture<InputStream> f = httpClient.go(request, handler);\ntry {\n  f.get(timeout, TimeUnit.MILLISECONDS);\n} catch (TimeoutException te) {\n  f.cancel(false); // avoid interrupting producer mid-put unless cancelling\n  throw te;\n}","handlingStrategy":"try-catch","validationCode":"// Ensure the consumer is alive before streaming large responses\nif (consumerClosed) {\n  responseFuture.cancel(false); // stop download without interrupting producer\n  return;\n}","typeGuard":null,"tryCatchPattern":"try {\n  queue.put(chunkStream);\n} catch (InterruptedException e) {\n  Thread.currentThread().interrupt();\n  throw new CancellationException(\"Response processing cancelled\");\n}","preventionTips":["Always consume or close the SequenceInputStream so the bounded queue drains","Use cancel(false) when you want the download to wind down without interrupts","Match HTTP client timeouts to expected response duration to avoid timeout-driven interrupts","Watch for consumer crashes that leave the queue full and the producer blocked"],"tags":["interrupt","http-client","queue","backpressure"],"backgroundTag":"thread-interrupted","analyzedSha":"9b90983fd291f26935af934383ce360473179e4d","analyzedAt":"2026-09-07T13:32:30.957Z","contentChangedAt":"2026-09-07T13:32:30.957Z","schemaVersion":2},"datasetVersion":"2026-09-17T15:17:12.973Z"}