apache/beam · error

broken stream: %v

Error message

broken stream: %v

What it means

This is a generated optimized reader for byte-slice keyed iteration (key []byte, value typex.X). It reads from the underlying element stream; on any read error other than io.EOF (which signals normal end of stream) it panics with 'broken stream', wrapping the cause. A broken stream means the element channel or reader backing the iteration failed mid-stream.

Source

Thrown at sdks/go/pkg/beam/core/runtime/exec/optimized/inputs.go:952

	}
	*key = elm.Elm.([]byte)
	*value = elm.Elm2.(typex.W)
	return true
}

func iterMakerByteSliceTypex_W(s exec.ReStream) exec.ReusableInput {
	ret := &iterNative{s: s}
	ret.fn = ret.readByteSliceTypex_W
	return ret
}

func (v *iterNative) readByteSliceTypex_X(key *[]byte, value *typex.X) bool {
	elm, err := v.cur.Read()
	if err != nil {
		if err == io.EOF {
			return false
		}
		panic(fmt.Sprintf("broken stream: %v", err))
	}
	*key = elm.Elm.([]byte)
	*value = elm.Elm2.(typex.X)
	return true
}

func iterMakerByteSliceTypex_X(s exec.ReStream) exec.ReusableInput {
	ret := &iterNative{s: s}
	ret.fn = ret.readByteSliceTypex_X
	return ret
}

func (v *iterNative) readByteSliceTypex_Y(key *[]byte, value *typex.Y) bool {
	elm, err := v.cur.Read()
	if err != nil {
		if err == io.EOF {
			return false
		}

View on GitHub (pinned to 12126d8942)

Solutions

1. Read the wrapped cause in the panic message to identify whether it is transport, decode, or runner-side.
2. Retry the bundle/work item — transient channel failures often succeed on retry.
3. Check runner (e.g. Dataflow/Flink) logs for the producing stage's health and any channel closures.
4. If caused by large side inputs, reduce side input size or switch to a different input pattern (e.g. co-group) to avoid streaming limits.
5. Upgrade the SDK/runner if the error is a known deserialization regression.

defer func() {
	if r := recover(); r != nil {
		if s, ok := r.(string); ok && strings.HasPrefix(s, "broken stream") {
			// surface wrapped cause and retry the work item via the runner's retry mechanism
			log.Printf("side input stream failed: %s", s)
		}
	}
}()

When it happens

Trigger: v.cur.Read() returns a non-EOF error during side-input iteration — e.g. the remote channel carrying side input data failed, decoding of an element failed, or the runner closed the stream prematurely.

Common situations: Network failures between runner and SDK harness while streaming side input data; runner OOM/crash truncating the side input stream; data corruption or deserialization errors on element decode.

Related errors

• broken stream: %v
• combine does not support side inputs
• side input closed
• invalid side pcollection: index %v
• values of %v cannot bind to %v

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