{"record":{"id":"4345df5c18344391","repo":"apache/beam","slug":"instruction-v-no-longer-processing-statemgr","errorCode":null,"errorMessage":"instruction %v no longer processing","messagePattern":"instruction (.+?) no longer processing","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/core/runtime/harness/statemgr.go","lineNumber":188,"sourceCode":"\t})\n\treturn wr, err\n}\n\n// GetSideInputCache returns a pointer to the SideInputCache being used by the SDK harness.\nfunc (s *ScopedStateReader) GetSideInputCache() exec.SideCache {\n\treturn s.cache\n}\n\nfunc (s *ScopedStateReader) openReader(ctx context.Context, id exec.StreamID, readerFn func(*StateChannel) *stateKeyReader) (*stateKeyReader, error) {\n\tch, err := s.open(ctx, id.Port)\n\tif err != nil {\n\t\treturn nil, err\n\t}\n\n\ts.mu.Lock()\n\tif s.closed {\n\t\ts.mu.Unlock()\n\t\treturn nil, errors.Errorf(\"instruction %v no longer processing\", s.instID)\n\t}\n\tret := readerFn(ch)\n\ts.mu.Unlock()\n\treturn ret, nil\n}\n\nfunc (s *ScopedStateReader) openWriter(ctx context.Context, id exec.StreamID, writerFn func(*StateChannel) *stateKeyWriter) (*stateKeyWriter, error) {\n\tch, err := s.open(ctx, id.Port)\n\tif err != nil {\n\t\treturn nil, err\n\t}\n\n\ts.mu.Lock()\n\tif s.closed {\n\t\ts.mu.Unlock()\n\t\treturn nil, errors.Errorf(\"instruction %v no longer processing\", s.instID)\n\t}\n\tret := writerFn(ch)","sourceCodeStart":170,"sourceCodeEnd":206,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/core/runtime/harness/statemgr.go#L170-L206","documentation":"ScopedStateReader.openReader in the Beam Go harness refuses to open a new side-input/user-state reader when the scoped state reader has already been closed for the current bundle instruction. Once an instruction's processing finishes, its state channels are torn down; opening a reader afterwards would target a dead instruction. The library throws this to prevent reads against state that will never be answered.","triggerScenarios":"Calling OpenIterableSideInput, OpenMultiMapSideInput, OpenIterable, or any Open*UserStateReader after the ScopedStateReader was closed (bundle ended or Close() invoked), typically via a DoFn continuing to touch user state or side inputs after the bundle context finished.","commonSituations":"DoFn goroutines outliving the bundle (FinishBundle or process cleanup racing with state access), reusing exec units across bundles after close, async/callback code firing after the harness closed the instruction's state scope.","solutions":["Ensure all state/side-input reads happen inside the ProcessElement/StartBundle lifecycle, before Close() is called on the state reader.","Fix goroutine leaks or deferred callbacks in DoFns that touch user state after the bundle ends.","Check bundle/instruction lifecycle handling in custom exec code; do not cache ScopedStateReader across bundles.","If caused by a Beam runtime bug (e.g. state access in teardown), report with the pipeline and upgrade to the latest Beam Go SDK version."],"exampleFix":"// before (goroutine outlives bundle)\ngo func() { it, _ := ctx.sideInput(ctx) }()\n// after: perform side-input reads synchronously inside ProcessElement\nit := s.side.Read(ctx, s.sideInput)","handlingStrategy":"try-catch","validationCode":"if s.IsClosed() { return fmt.Errorf(\"skip: state scope closed for instruction\") }","typeGuard":null,"tryCatchPattern":"iter, err := reader.OpenIterableSideInput(ctx, id, w)\nif err != nil {\n    if strings.Contains(err.Error(), \"no longer processing\") {\n        return nil // bundle ended; abort read gracefully\n    }\n    return err\n}","preventionTips":["Read side inputs only inside ProcessElement","No goroutines touching state after FinishBundle","Do not cache state readers across bundles"],"tags":["go","beam","state-management","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-20T03:17:13.778Z"}