What causes "decodeReStream opened twice" in apache/beam?

Open the ReStream exactly once and keep the returned Stream for all reads.

decodeReStream opened twice

Error message

decodeReStream opened twice

What it means

singleUseReStream is a one-shot in-memory ReStream: Open() hands out the underlying reader exactly once and nils it out. A second Open() call finds n.r == nil and returns this error, because the stream cannot be replayed from the start.

Solutions

Open the ReStream exactly once and keep the returned Stream for all reads.
If the data must be consumed multiple times, wrap the source so each consumer gets a fresh singleUseReStream.
Refactor the pipeline so the decoded element is shared instead of the stream.

Defensive patterns

Strategy: try-catch

Validate before calling

// Track whether the stream was already opened
if stream.opened { return errors.New("singleUseReStream already consumed") }

Type guard

func isOpen(n *exec.singleUseReStream) bool { return n != nil /* and not yet opened; inspect r via package-local code */ }

Try / catch

s, err := rStream.Open()
if err != nil {
    // stream already consumed: re-materialize or obtain a fresh ReStream
    return fmt.Errorf("stream unusable: %w", err)
}

Prevention

Open each single-use ReStream exactly once, in one place
Pass decoded values, not streams, to multiple consumers
Never cache the ReStream for later re-reading

When it happens

Trigger: Calling Open() twice on the same singleUseReStream, e.g. reading an element's value in two different exec nodes or retrying a consumption path that already opened the stream.

Common situations: Custom exec transform code that caches a FullValue and re-reads it; misconfiguring a dedup/caching layer that opens streams twice; debugging the exec package with repeated Open calls.

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/61b21ae177f8bf2a. Report an issue: GitHub.

Appendix: source

Thrown at sdks/go/pkg/beam/core/runtime/exec/fullvalue.go:208

		}
		ret = append(ret, *elm)
	}
}

// singleUseReStream is a decode on demand ReStream.
// Can only produce a single Stream because it consumes the reader.
// Must not be used for streams that might be re-iterated, causing Open
// to be called twice.
type singleUseReStream struct {
	r    io.Reader
	d    ElementDecoder
	size int // The number of elements in this stream.
}

// Open returns the Stream from the start of the in-memory reader. Returns error if called twice.
func (n *singleUseReStream) Open() (Stream, error) {
	if n.r == nil {
		return nil, errors.New("decodeReStream opened twice")
	}
	ret := &decodeStream{r: n.r, d: n.d, size: n.size}
	n.r = nil
	n.d = nil
	return ret, nil
}

// decodeStream is a decode on demand Stream, that decodes size elements from the provided
// io.Reader.
type decodeStream struct {
	r          io.Reader
	d          ElementDecoder
	next, size int
	ret        FullValue
}

// Close causes subsequent calls to Read to return io.EOF, and drains the remaining element count
// from the reader.

View on GitHub (pinned to 12126d8942)