{"record":{"id":"6824d4a8724b0040","repo":"apache/beam","slug":"instruction-v-no-longer-processing","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/datamgr.go","lineNumber":86,"sourceCode":"\t\treturn nil, err\n\t}\n\treturn ch.OpenElementChan(ctx, id.PtransformID, s.instID, expectedTimerTransforms)\n}\n\n// OpenTimerWrite opens an io.WriteCloser on the given stream to write timers\nfunc (s *ScopedDataManager) OpenTimerWrite(ctx context.Context, id exec.StreamID, family string) (io.WriteCloser, error) {\n\tch, err := s.open(ctx, id.Port)\n\tif err != nil {\n\t\treturn nil, err\n\t}\n\treturn ch.OpenTimerWrite(ctx, id.PtransformID, s.instID, family), nil\n}\n\nfunc (s *ScopedDataManager) open(ctx context.Context, port exec.Port) (*DataChannel, error) {\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\ts.openPorts = append(s.openPorts, port)\n\tlocal := s.mgr\n\ts.mu.Unlock()\n\n\treturn local.Open(ctx, port) // don't hold lock over potentially slow operation\n}\n\n// Close prevents new IO for this instruction.\nfunc (s *ScopedDataManager) Close() error {\n\ts.mu.Lock()\n\tdefer s.mu.Unlock()\n\ts.closed = true\n\terr := s.mgr.closeInstruction(s.instID, s.openPorts)\n\ts.mgr = nil\n\treturn err\n}\n","sourceCodeStart":68,"sourceCodeEnd":104,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/core/runtime/harness/datamgr.go#L68-L104","documentation":"ScopedDataManager.open refuses to open a new data channel port when the manager for this instruction has already been closed. The Beam Go harness closes the ScopedDataManager once an instruction (bundle or process-bundle request) finishes processing; any late attempt to open a port afterwards is invalid state. This prevents writes to a data plane connection that no longer has a serving instruction.","triggerScenarios":"Calling OpenWrite, OpenElementChan, or OpenTimerWrite after ScopedDataManager.Close() was invoked for the instruction ID; a runner issuing additional port opens after the bundle completed; concurrent open racing a close between checking and opening.","commonSituations":"Bundle already finished but the SDK still holds a DoFn that lazily opens a side-output or timer writer; harness shutdown ordering issues where logging/emitter goroutines outlive the instruction; retry logic that re-opens channels without recreating the ScopedDataManager.","solutions":["Ensure all channel opens (OpenWrite/OpenElementChan/OpenTimerWrite) happen before the instruction's processing completes; do not retain writers beyond bundle lifetime","Check that the instructionID used for the ScopedDataManager matches the currently processing instruction; do not reuse a closed manager","In concurrent code, serialize opens with Close so no open lands after shutdown","If this appears during pipeline shutdown, verify the runner is not sending extra bundle requests with a completed instruction ID"],"exampleFix":"// before\nw := mgr.OpenWrite(ctx, port) // may fail if instruction already closed\n// after\nif err := mgr.EnsureOpen(ctx, instructionID); err != nil {\n  return fmt.Errorf(\"instruction %s no longer processing; cannot open port: %w\", instructionID, err)\n}\nw := mgr.OpenWrite(ctx, port)","handlingStrategy":"try-catch","validationCode":"if mgr == nil || mgr.IsClosed() {\n  return fmt.Errorf(\"scoped datamanager for %s already closed; refusing to open port\", instID)\n}","typeGuard":"func openable(m *ScopedDataManager) bool {\n  m.mu.Lock(); defer m.mu.Unlock()\n  return !m.closed\n}","tryCatchPattern":"ch, err := mgr.OpenWrite(ctx, port)\nif err != nil && strings.Contains(err.Error(), \"no longer processing\") {\n  // bundle finished; drop late writer or re-open with a fresh instruction\n  return ErrInstructionClosed\n}","preventionTips":["Open all writers/reader channels at bundle start, not lazily in DoFns","Never cache ScopedDataManager across bundle boundaries","Use sync.Once or lifecycle hooks so opens cannot race Close"],"tags":["go","apache-beam","state","data-channel"],"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"}