opening a reader %v on a closed channel. Original error: %w

Error message

opening a reader %v on a closed channel. Original error: %w

What it means

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.

Source

Thrown at sdks/go/pkg/beam/core/runtime/harness/datamgr.go:311

	c.cancelFn() // A context.CancelFunc is threadsafe and indempotent.
	if c.forceRecreate != nil {
		c.forceRecreate(c.id, err)
		c.forceRecreate = nil
	}
}

// OpenWrite returns an io.WriteCloser of the data elements for the given instruction and ptransform.
func (c *DataChannel) OpenWrite(ctx context.Context, ptransformID string, instID instructionID) io.WriteCloser {
	return c.makeWriter(ctx, clientID{ptransformID: ptransformID, instID: instID})
}

// OpenElementChan returns a channel of typex.Elements for the given instruction and ptransform.
func (c *DataChannel) OpenElementChan(ctx context.Context, ptransformID string, instID instructionID, expectedTimerTransforms []string) (<-chan exec.Elements, error) {
	c.mu.Lock()
	defer c.mu.Unlock()
	cid := clientID{ptransformID: ptransformID, instID: instID}
	if c.readErr != nil {
		return nil, fmt.Errorf("opening a reader %v on a closed channel. Original error: %w", cid, c.readErr)
	}
	return c.makeChannel(true, cid, expectedTimerTransforms...).ch, nil
}

// makeChannel creates a channel of exec.Elements. It expects to be called while c.mu is held.
func (c *DataChannel) makeChannel(fromSource bool, id clientID, additionalTransforms ...string) *elementsChan {
	if ec, ok := c.channels[id.instID]; ok {
		ec.mu.Lock()
		defer ec.mu.Unlock()
		if fromSource {
			ec.want = (1 + int32(len(additionalTransforms)))
		}
		if _, ok := c.endedInstructions[id.instID]; ok || (ec.want > 0 && ec.want == ec.got) {
			atomic.StoreUint32(&ec.closed, 1)
			close(ec.ch)
		}
		return ec
	}

View on GitHub (pinned to 12126d8942)

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.

Example fix

// before
ch, err := ch0.OpenElementChan(ctx, pid, closedInstID, nil) // inst already closed
// after
if err := dc.CheckErr(); err != nil { return err } // bail out on closed channel
ch, err := dc.OpenElementChan(ctx, pid, newInstID, nil)

Defensive patterns

Strategy: try-catch

Validate before calling

if err := dc.CheckErr(); err != nil { // channel already closed
	return err
}

Try / catch

ch, err := dc.OpenElementChan(ctx, pid, inst, nil)
if err != nil {
	var oe error
	if errors.As(err, &oe) { /* inspect wrapped original close error */ }
	return err // abort bundle; runner will retry with a fresh instruction
}

Prevention

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.

When it happens

Trigger: 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.

Common situations: 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.

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

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