apache/beam · error

broken stream: %v

Error message

broken stream: %v

What it means

iter1[T].invoke reads the next element from the side-input/reiteration stream; EOF cleanly ends iteration (returns false), but any other read error triggers panic("broken stream: %v", err). This signals the element stream (typically a gRPC-backed reader between stages) failed mid-iteration, so iteration cannot continue safely.

Source

Thrown at sdks/go/pkg/beam/register/iter.go:63

func (v *iter1[T]) Value() any {
	return v.invoke
}

func (v *iter1[T]) Reset() error {
	if err := v.cur.Close(); err != nil {
		return err
	}
	v.cur = nil
	return nil
}

func (v *iter1[T]) invoke(value *T) bool {
	elm, err := v.cur.Read()
	if err != nil {
		if err == io.EOF {
			return false
		}
		panic(fmt.Sprintf("broken stream: %v", err))
	}
	*value = elm.Elm.(T)
	return true
}

type iter2[T1, T2 any] struct {
	s exec.ReStream

	// cur is the "current" stream, if any.
	cur exec.Stream
}

func (v *iter2[T1, T2]) Init() error {
	cur, err := v.s.Open()
	if err != nil {
		return err
	}
	v.cur = cur

View on GitHub (pinned to 12126d8942)

Solutions

1. Check the wrapped error for the transport/decode failure and retry the pipeline/workers.
2. Reduce side-input size (filter earlier) or switch to a CoGBK/GBK join to avoid huge side inputs.
3. Verify runner/SDK version compatibility (mismatched beam versions can break the wire protocol).
4. If transient network issues, increase runner-level retry/network timeouts.

Example fix

// before
for v.Range(&x) {
    process(x) // panics mid-iteration on broken stream
}

// after
// runner-level: pin matching SDK/runner versions and shrink side inputs
side := beam.SideInput{Input: beam.CombinePerKey(sum, bigPColl)}
small := beam.ParDo(s, filteredFn, smallPColl) // filter before side input

// no pre-call validation possible; ensure side inputs are small and versions match

defer func() {
    if r := recover(); r != nil {
        if s, ok := r.(string); ok && strings.HasPrefix(s, "broken stream") {
            log.Printf("side input stream broken: %s", s) // trigger runner retry
        }
    }
}()

When it happens

Trigger: Iterating a side input or grouped input where cur.Read() returns a non-EOF error: the remote data channel broke, deserialization failed, or the upstream writer died mid-stream.

Common situations: Worker-to-worker gRPC side-input reads over unstable networks; runner-side bugs truncating streams; large side inputs hitting connection/data limits in Dataflow or Flink runners.

Related errors

• broken stream
• broken stream: %v
• session windowing is not supported for side inputs
• broken stream: %v
• broken stream: %v

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