apache/beam · critical

broken stream

Error message

broken stream: %v

What it means

This panic comes from the generated iterator shim readTypex۰T in the Beam Go SDK debug/x package. It wraps a single-element source whose underlying stream read returned a non-EOF error; because a partially broken stream cannot be recovered inside a DoFn iterator, the shim panics with the underlying cause. It is thrown only when the read error is anything other than io.EOF (which signals normal end of iteration).

Solutions

1. Inspect the wrapped %v cause in the panic message to identify the transport/source failure (usually a gRPC or connection error).
2. Check runner worker logs for harness/runner disconnections and fix network stability or increase timeouts.
3. Retry the pipeline; for streaming jobs configure runner-level restart/checkpointing so a broken channel recovers.
4. If reproducible locally, run with a stable direct runner or fixed TestStream data to isolate whether the source or the network is at fault.

// Go panics cannot be caught by error returns; wrap the DoFn body if you must contain it:
func safeRead(v *iterNative, value *beam.T) (ok bool) {
    defer func() {
        if r := recover(); r != nil {
            log.Printf("stream read failed: %v", r)
            ok = false
        }
    }()
    return v.readTypex۰T(value)
}

When it happens

Trigger: Calling v.cur.Read() on the element stream inside a ParDo whose iterator is consumed via range, and the underlying source (e.g. a teststream, data channel, or side input reader) returns an error other than io.EOF — e.g. connection dropped to the runner, corrupted gRPC byte stream, or harness terminated mid-stream.

Common situations: Runner (Flink/Spark/Dataflow) worker connection drops during a streaming job; a TestStream or debug source feeding x/debug shims fails mid-iteration; network partitions between harness and runner causing the element reader to return a transport error.

Related errors

• broken stream
• broken stream
• broken stream
• AfterProcessingTime trigger set without a delay or…
• At least one subtrigger required for composite triggers.

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

Appendix: source

Thrown at sdks/go/pkg/beam/x/debug/debug.shims.go:365

		return err
	}
	v.cur = nil
	return nil
}

func iterMakerTypex۰T(s exec.ReStream) exec.ReusableInput {
	ret := &iterNative{s: s}
	ret.fn = ret.readTypex۰T
	return ret
}

func (v *iterNative) readTypex۰T(value *beam.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.(beam.T)
	return true
}

func iterMakerTypex۰XTypex۰Y(s exec.ReStream) exec.ReusableInput {
	ret := &iterNative{s: s}
	ret.fn = ret.readTypex۰XTypex۰Y
	return ret
}

func (v *iterNative) readTypex۰XTypex۰Y(key *beam.X, value *beam.Y) bool {
	elm, err := v.cur.Read()
	if err != nil {
		if err == io.EOF {
			return false
		}
		panic(fmt.Sprintf("broken stream: %v", err))

View on GitHub (pinned to 12126d8942)