panic in stage.Execute bundle processing goroutine: %v, stag

Error message

panic in stage.Execute bundle processing goroutine: %v, stage: %+v,stackTrace:
%s

What it means

stage.Execute recovers any panic that occurs in the bundle-processing goroutine and converts it into this error including the panic value, the stage, and a stack trace. The bundle fails (rather than crashing the whole runner process) so the job can report the failure.

Source

Thrown at sdks/go/pkg/beam/runners/prism/internal/stage.go:132

	for k, v := range s.OutputsToCoders {
		outAttrs = append(outAttrs, slog.Any(k, v))
	}
	return slog.GroupValue(
		slog.String("ID", s.ID),
		slog.Any("transforms", s.transforms),
		slog.Any("inputInfo", s.inputInfo),
		slog.Group("outputInfo", outAttrs...),
	)
}

func (s *stage) Execute(ctx context.Context, j *jobservices.Job, wk *worker.W, comps *pipepb.Components, em *engine.ElementManager, rb engine.RunBundle) (err error) {
	if s.baseProgTick.Load() == nil {
		s.baseProgTick.Store(minimumProgTick)
	}
	defer func() {
		// Convert execution panics to errors to fail the bundle.
		if e := recover(); e != nil {
			err = fmt.Errorf("panic in stage.Execute bundle processing goroutine: %v, stage: %+v,stackTrace:\n%s", e, s, debug.Stack())
		}
	}()
	slog.Debug("Execute: starting bundle", "bundle", rb)

	var b *worker.B
	initialState := em.StateForBundle(rb)
	var dataReady <-chan struct{}
	switch s.envID {
	case "": // Runner Transforms
		if len(s.transforms) != 1 {
			panic(fmt.Sprintf("unexpected number of runner transforms, want 1: %+v", s))
		}
		tid := s.transforms[0]
		// Runner transforms are processed immeadiately.
		b = s.exe.ExecuteTransform(s.ID, tid, comps.GetTransforms()[tid], comps, rb.Watermark, em.InputForBundle(rb, s.inputInfo))
		b.InstID = rb.BundleID
		slog.Debug("Execute: runner transform", "bundle", rb, slog.String("tid", tid))

View on GitHub (pinned to 12126d8942)

Solutions

Read the included stackTrace to locate the panicking prism code
Reproduce with the specific bundle data captured in logs
File a Beam issue with the full stack trace if the panic originates in prism
Pin to a Beam version where the panic is fixed

Defensive patterns

Strategy: try-catch

Try / catch

if err := stage.Execute(ctx, wk, comps, s, rb); err != nil {
    if strings.HasPrefix(err.Error(), "panic in stage.Execute") {
        slog.Error("stage panic", "stack", extractStack(err))
    }
    return err
}

Prevention

Capture full bundle inputs when a panic occurs for reproduction
Keep prism at a patched Beam release
Report panics with stack traces to the Beam project

When it happens

Trigger: Any panic inside bundle execution for a stage — nil dereference in prism's bundle processing, a bad coder causing index panics, or unrecovered panics from internal state handling.

Common situations: Processing unexpected data that triggers unbounded assumptions; concurrent state access bugs; runner bugs around timers/state on new SDK features.

Related errors

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