apache/beam · warning · RuntimeException

Interrupted while waiting for space in buffer

Error message

Interrupted while waiting for space in buffer

What it means

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.

Source

Thrown at sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/AsyncWrapper.java:518

  }

  // Schedule an element to the thread pool, retries with backoff if the buffer is full.
  private void scheduleItem(KV<K, InputT> element, BoundedWindow window, Instant timestamp) {
    boolean done = false;
    long sleepTime = INITIAL_BACKOFF_SLEEP_MS;
    long totalSleep = 0;
    long timeoutMs = timeout.getMillis();

    while (!done && totalSleep < timeoutMs) {
      done = scheduleIfRoom(element, window, timestamp, false);
      if (!done) {
        long sleep = Math.min(maxWaitTime.getMillis(), sleepTime);
        logBackpressure(element, sleep, totalSleep);
        try {
          Thread.sleep(sleep);
        } catch (InterruptedException e) {
          Thread.currentThread().interrupt();
          throw new RuntimeException("Interrupted while waiting for space in buffer", e);
        }

        // Prevents long overflow possibility
        if (sleepTime < maxWaitTime.getMillis()) {
          sleepTime *= 2;
        }

        totalSleep += sleep;
      }
    }
    // Timeout: element skips JVM pool but stays in BagState for timer to reschedule later.
  }

  // Uses hashcode based jitter instead of random for deterministic rescheduling
  // Satisfies lint check
  private Instant nextTimeToFire(@Nullable K key) {
    long seed = (key == null) ? 0 : key.hashCode();
    double fractionalOffset = Math.abs(seed % (long) HASH_MODULO_LIMIT) / HASH_MODULO_LIMIT;

View on GitHub (pinned to 12126d8942)

Solutions

1. Treat this as a shutdown signal: check the pipeline/runner status before assuming a code bug
2. Avoid blocking calls inside the async fn so the buffer does not fill and the thread does not sit in the sleep loop
3. Catch RuntimeException at the pipeline-submission layer and check the interrupted cause to confirm teardown
4. Retry the pipeline run if the interruption was an accidental cancellation

Example fix

// before: blocking inside fn makes buffer fill
fn: element -> blockingRpc(element)
// after: non-blocking async client
fn: element -> asyncClient.sendAsync(element)

if (pipeline.getState().isTerminal()) throw new IllegalStateException("Pipeline already stopped; not starting async work");

try {
  pipeline.run().waitUntilFinish();
} catch (RuntimeException e) {
  if (e.getCause() instanceof InterruptedException) {
    LOG.warn("Async transform interrupted during shutdown; safe to re-run");
  } else { throw e; }
}

When it happens

Trigger: 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.

Common situations: 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.

Related errors

• executor service was interrupted
• Interrupted closing FlightClient
• Timing number 0b" + timingNumber.toString(2) + " has more th
• No proto encoding for PaneInfoCoder, always part of Windowed
• Illegal access to pipeline after visitor traversal was compl

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/dd9bf5ba876e9e73. Report an issue: GitHub.