{"record":{"id":"95018a8cb5c6c676","repo":"apache/beam","slug":"opening-a-reader-v-on-a-closed-channel-original-error-w","errorCode":null,"errorMessage":"opening a reader %v on a closed channel. Original error: %w","messagePattern":"opening a reader (.+?) on a closed channel\\. Original error: %w","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/core/runtime/harness/datamgr.go","lineNumber":311,"sourceCode":"\tc.cancelFn() // A context.CancelFunc is threadsafe and indempotent.\n\tif c.forceRecreate != nil {\n\t\tc.forceRecreate(c.id, err)\n\t\tc.forceRecreate = nil\n\t}\n}\n\n// OpenWrite returns an io.WriteCloser of the data elements for the given instruction and ptransform.\nfunc (c *DataChannel) OpenWrite(ctx context.Context, ptransformID string, instID instructionID) io.WriteCloser {\n\treturn c.makeWriter(ctx, clientID{ptransformID: ptransformID, instID: instID})\n}\n\n// OpenElementChan returns a channel of typex.Elements for the given instruction and ptransform.\nfunc (c *DataChannel) OpenElementChan(ctx context.Context, ptransformID string, instID instructionID, expectedTimerTransforms []string) (<-chan exec.Elements, error) {\n\tc.mu.Lock()\n\tdefer c.mu.Unlock()\n\tcid := clientID{ptransformID: ptransformID, instID: instID}\n\tif c.readErr != nil {\n\t\treturn nil, fmt.Errorf(\"opening a reader %v on a closed channel. Original error: %w\", cid, c.readErr)\n\t}\n\treturn c.makeChannel(true, cid, expectedTimerTransforms...).ch, nil\n}\n\n// makeChannel creates a channel of exec.Elements. It expects to be called while c.mu is held.\nfunc (c *DataChannel) makeChannel(fromSource bool, id clientID, additionalTransforms ...string) *elementsChan {\n\tif ec, ok := c.channels[id.instID]; ok {\n\t\tec.mu.Lock()\n\t\tdefer ec.mu.Unlock()\n\t\tif fromSource {\n\t\t\tec.want = (1 + int32(len(additionalTransforms)))\n\t\t}\n\t\tif _, ok := c.endedInstructions[id.instID]; ok || (ec.want > 0 && ec.want == ec.got) {\n\t\t\tatomic.StoreUint32(&ec.closed, 1)\n\t\t\tclose(ec.ch)\n\t\t}\n\t\treturn ec\n\t}","sourceCodeStart":293,"sourceCodeEnd":329,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/core/runtime/harness/datamgr.go#L293-L329","documentation":"OpenElementChan creates an element-reading channel for a given instruction on a DataChannel. If the channel was previously closed with an error (c.readErr set), opening a new reader is rejected and this error is returned, embedding the clientID (ptransformID/instID) and the original close error.","triggerScenarios":"Calling OpenElementChan after the data channel was closed/failed — typically after a prior read error, instruction cancellation, or data manager shutdown; seen in tests like TestDataChannelRemoveInstruction_dataAfterClose.","commonSituations":"Bundles attempting to read side inputs or sources after a stream failure or after the instruction's data was removed/closed; network drops between runner and SDK worker.","solutions":["Inspect the wrapped original error (%w) to find why the channel closed — fix that root cause first.","Do not re-open readers for instructions whose data was removed; issue a fresh instruction.","Ensure the runner keeps the data stream open until all expected readers complete.","Check for ordering bugs where reads happen after explicit close/remove of the channel."],"exampleFix":"// before\nch, err := ch0.OpenElementChan(ctx, pid, closedInstID, nil) // inst already closed\n// after\nif err := dc.CheckErr(); err != nil { return err } // bail out on closed channel\nch, err := dc.OpenElementChan(ctx, pid, newInstID, nil)","handlingStrategy":"try-catch","validationCode":"if err := dc.CheckErr(); err != nil { // channel already closed\n\treturn err\n}","typeGuard":null,"tryCatchPattern":"ch, err := dc.OpenElementChan(ctx, pid, inst, nil)\nif err != nil {\n\tvar oe error\n\tif errors.As(err, &oe) { /* inspect wrapped original close error */ }\n\treturn err // abort bundle; runner will retry with a fresh instruction\n}","preventionTips":["Never reuse instruction IDs after data removal/close.","Check channel error state before opening new readers.","Let the runner handle recovery instead of re-opening failed channels."],"tags":["data-channel","closed-channel","state","beam","go"],"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"}