{"record":{"id":"dd9bf5ba876e9e73","repo":"apache/beam","slug":"interrupted-while-waiting-for-space-in-buffer","errorCode":null,"errorMessage":"Interrupted while waiting for space in buffer","messagePattern":"Interrupted while waiting for space in buffer","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"warning","filePath":"sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/AsyncWrapper.java","lineNumber":518,"sourceCode":"  }\n\n  // Schedule an element to the thread pool, retries with backoff if the buffer is full.\n  private void scheduleItem(KV<K, InputT> element, BoundedWindow window, Instant timestamp) {\n    boolean done = false;\n    long sleepTime = INITIAL_BACKOFF_SLEEP_MS;\n    long totalSleep = 0;\n    long timeoutMs = timeout.getMillis();\n\n    while (!done && totalSleep < timeoutMs) {\n      done = scheduleIfRoom(element, window, timestamp, false);\n      if (!done) {\n        long sleep = Math.min(maxWaitTime.getMillis(), sleepTime);\n        logBackpressure(element, sleep, totalSleep);\n        try {\n          Thread.sleep(sleep);\n        } catch (InterruptedException e) {\n          Thread.currentThread().interrupt();\n          throw new RuntimeException(\"Interrupted while waiting for space in buffer\", e);\n        }\n\n        // Prevents long overflow possibility\n        if (sleepTime < maxWaitTime.getMillis()) {\n          sleepTime *= 2;\n        }\n\n        totalSleep += sleep;\n      }\n    }\n    // Timeout: element skips JVM pool but stays in BagState for timer to reschedule later.\n  }\n\n  // Uses hashcode based jitter instead of random for deterministic rescheduling\n  // Satisfies lint check\n  private Instant nextTimeToFire(@Nullable K key) {\n    long seed = (key == null) ? 0 : key.hashCode();\n    double fractionalOffset = Math.abs(seed % (long) HASH_MODULO_LIMIT) / HASH_MODULO_LIMIT;","sourceCodeStart":500,"sourceCodeEnd":536,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/AsyncWrapper.java#L500-L536","documentation":"AsyncJoin/AsyncWrapper applies backpressure by sleeping when its bounded in-flight buffer is full. If the worker thread's sleep is interrupted (typically pipeline teardown or a cancellation signal), it restores the interrupt flag and rethrows as a RuntimeException so the DoFn fails fast instead of silently dropping the element.","triggerScenarios":"Thread.sleep inside scheduleItem's backpressure loop (buffer full) is interrupted by Thread.interrupt(), usually during pipeline shutdown, runner cancellation, or a watchdog killing a stuck worker.","commonSituations":"Cancelling a Dataflow/Flink/Spark job mid-run; runner killing a worker thread that exceeded a time limit; JVM shutdown hooks interrupting pipeline threads; user code in an async transform blocking the bundle thread so the buffer fills and then teardown interrupts it.","solutions":["Treat this as a shutdown signal: check the pipeline/runner status before assuming a code bug","Avoid blocking calls inside the async fn so the buffer does not fill and the thread does not sit in the sleep loop","Catch RuntimeException at the pipeline-submission layer and check the interrupted cause to confirm teardown","Retry the pipeline run if the interruption was an accidental cancellation"],"exampleFix":"// before: blocking inside fn makes buffer fill\nfn: element -> blockingRpc(element)\n// after: non-blocking async client\nfn: element -> asyncClient.sendAsync(element)","handlingStrategy":"retry","validationCode":"if (pipeline.getState().isTerminal()) throw new IllegalStateException(\"Pipeline already stopped; not starting async work\");","typeGuard":null,"tryCatchPattern":"try {\n  pipeline.run().waitUntilFinish();\n} catch (RuntimeException e) {\n  if (e.getCause() instanceof InterruptedException) {\n    LOG.warn(\"Async transform interrupted during shutdown; safe to re-run\");\n  } else { throw e; }\n}","preventionTips":["Avoid blocking calls inside async fns so backpressure sleeps are short","Check pipeline/job state before and during long async operations","Configure timeouts on async clients so bundles finish before runner watchdogs fire"],"tags":["java","backpressure","interruption","pipeline-teardown"],"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"}