apache/beam · error · Error

State stream is closed.

Error message

State stream is closed.

What it means

The worker's RemoteGrpcStateClient/getState path performs state requests over the state stream. If the handler was closed (bundle finished or connection torn down) any new getState call throws 'State stream is closed.' — the library does this because requests can no longer be answered on a closed stream.

Source

Thrown at sdks/typescript/src/apache_beam/worker/state.ts:272

      cb!(response);
    });
    this.stateChannel.on("error", (error) => {
      this.error = error;
    });
  }

  close() {
    this.closed = true;
    this.stateChannel.end();
  }

  getState<T>(
    instructionId: string,
    stateKey: fnApi.StateKey,
    decode: (data: Uint8Array) => T,
  ): MaybePromise<T> {
    if (this.closed) {
      throw new Error("State stream is closed.");
    } else if (this.error) {
      throw this.error;
    }

    const this_ = this;

    // Not inlined as it may need to be called recursively to handle
    // continuation tokens.
    function responseCallback(
      resolve,
      reject,
      prevChunks: Uint8Array[] = [],
    ): (response: fnApi.StateResponse) => void {
      return (response) => {
        if (this_.error) {
          reject(this_.error);
          return;
        }

View on GitHub (pinned to 12126d8942)

Solutions

1. Ensure all state reads happen during the bundle that owns the provider, before it closes.
2. Check this.closed/this.error before calling getState if you hold a reference, and skip or re-obtain a provider otherwise.
3. Re-architect late async work so it completes (or is cancelled) before bundle teardown.
4. If hitting this during normal pipeline runs, report the race to Beam with the job logs.

Example fix

// before
// await new Promise(r => setTimeout(r)); state = provider.getState(...) // stream already closed
// after
// if (!providerClosed) { state = provider.getState(...) } // or await pending state reads before bundle end

// caller-side: only read state while the bundle is live
if (handler.closed || handler.error) {
  throw new Error('Skipping state read: handler closed');
}

const canReadState = (h: {closed: boolean; error?: unknown}): boolean => !h.closed && !h.error;

try {
  const v = await provider.getState(instructionId, key, decode);
} catch (e) {
  if (e.message === 'State stream is closed.') {
    // bundle ended; re-fetch provider or abort the late callback
    return;
  }
  throw e;
}

When it happens

Trigger: Calling getStateProvider().getState(...) (directly or via DoFn state access) after the bundle/instruction completed and close() was called on the state handler, or reusing a cached state provider after the run ended.

Common situations: Asynchronous callbacks (timers, promise continuations) that outlive the bundle and try to read side inputs/state late; client code holding a state provider across process() calls.

Understand the failure class

Background: "Invalid state transition" errors: "status must be X, actually Y", "already rejected/charging/uninstalled", "cannot ... while running" — what they mean when a library rejects your call — this error's family across 31 libraries.

Related errors

• Not currently processing a bundle.
• Unknown context parameter: ${param.parDoParamName}
• Unsupported window mapping fn: ${sideInput.windowMappingFn.u
• Unimplemented access pattern: ${accessPattern}
• Illegal access to pipeline after visitor traversal was compl

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