{"record":{"id":"d1b5376cbd24b505","repo":"apache/beam","slug":"state-stream-is-closed","errorCode":null,"errorMessage":"State stream is closed.","messagePattern":"State stream is closed\\.","errorType":"exception","errorClass":"Error","httpStatus":null,"severity":"error","filePath":"sdks/typescript/src/apache_beam/worker/state.ts","lineNumber":272,"sourceCode":"      cb!(response);\n    });\n    this.stateChannel.on(\"error\", (error) => {\n      this.error = error;\n    });\n  }\n\n  close() {\n    this.closed = true;\n    this.stateChannel.end();\n  }\n\n  getState<T>(\n    instructionId: string,\n    stateKey: fnApi.StateKey,\n    decode: (data: Uint8Array) => T,\n  ): MaybePromise<T> {\n    if (this.closed) {\n      throw new Error(\"State stream is closed.\");\n    } else if (this.error) {\n      throw this.error;\n    }\n\n    const this_ = this;\n\n    // Not inlined as it may need to be called recursively to handle\n    // continuation tokens.\n    function responseCallback(\n      resolve,\n      reject,\n      prevChunks: Uint8Array[] = [],\n    ): (response: fnApi.StateResponse) => void {\n      return (response) => {\n        if (this_.error) {\n          reject(this_.error);\n          return;\n        }","sourceCodeStart":254,"sourceCodeEnd":290,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/typescript/src/apache_beam/worker/state.ts#L254-L290","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Ensure all state reads happen during the bundle that owns the provider, before it closes.","Check this.closed/this.error before calling getState if you hold a reference, and skip or re-obtain a provider otherwise.","Re-architect late async work so it completes (or is cancelled) before bundle teardown.","If hitting this during normal pipeline runs, report the race to Beam with the job logs."],"exampleFix":"// before\n// await new Promise(r => setTimeout(r)); state = provider.getState(...) // stream already closed\n// after\n// if (!providerClosed) { state = provider.getState(...) } // or await pending state reads before bundle end","handlingStrategy":"try-catch","validationCode":"// caller-side: only read state while the bundle is live\nif (handler.closed || handler.error) {\n  throw new Error('Skipping state read: handler closed');\n}","typeGuard":"const canReadState = (h: {closed: boolean; error?: unknown}): boolean => !h.closed && !h.error;","tryCatchPattern":"try {\n  const v = await provider.getState(instructionId, key, decode);\n} catch (e) {\n  if (e.message === 'State stream is closed.') {\n    // bundle ended; re-fetch provider or abort the late callback\n    return;\n  }\n  throw e;\n}","preventionTips":["Never retain state providers beyond the bundle that created them","Await/cancel async work before bundle teardown","Check handler.closed before issuing state calls"],"tags":["beam","worker","state","grpc","lifecycle"],"backgroundTag":"invalid-state-transition","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}