panic in ElementManager.Bundles watermark evaluation gorouti

Error message

panic in ElementManager.Bundles watermark evaluation goroutine: %v
%v

What it means

prism's ElementManager runs watermark evaluation in a dedicated goroutine that generates bundles. If that goroutine panics, the recover() converts the panic into an error passed to the job's cancel function, failing and cancelling the whole job with the panic value and stack trace.

Source

Thrown at sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go:410

		slog.Debug("no more pending elements: terminating pipeline")
		cancelFn(fmt.Errorf("elementManager out of elements, cleaning up"))
		// Ensure the watermark evaluation goroutine exits by locking the mutex
		// before broadcasting, preventing a lost wake-up signal.
		em.refreshCond.L.Lock()
		em.refreshCond.Broadcast()
		em.refreshCond.L.Unlock()
	}()
	// Watermark evaluation goroutine.
	go func() {
		// We should defer closing of the channel first, so that when a panic happens,
		// we will handle the panic and trigger a job failure BEFORE the job is
		// prematurely marked as done.
		defer close(runStageCh)
		defer func() {
			// In case of panics in bundle generation, fail and cancel the job.
			if e := recover(); e != nil {
				slog.Error("panic in ElementManager.Bundles watermark evaluation goroutine", "error", e, "traceback", string(debug.Stack()))
				upstreamCancelFn(fmt.Errorf("panic in ElementManager.Bundles watermark evaluation goroutine: %v\n%v", e, string(debug.Stack())))
			}
		}()

		for {
			em.refreshCond.L.Lock()
			// Check if processing time has advanced before the wait loop.
			emNow := em.processingTimeNow()
			changedByProcessingTime := em.processTimeEvents.AdvanceTo(emNow)
			em.changedStages.merge(changedByProcessingTime)

			// If there are no changed stages, ready processing time events,
			// or injected bundles available, we wait until there are.
			for len(em.changedStages)+len(changedByProcessingTime)+len(em.injectedBundles) == 0 {
				// Check to see if we must exit
				select {
				case <-ctx.Done():
					em.refreshCond.L.Unlock()
					return

View on GitHub (pinned to 12126d8942)

Solutions

Inspect the stack trace in the error for the panic site inside elementmanager.go
Reduce the pipeline to the minimal stage that triggers the panic and file a Beam issue with the trace
Try a newer Beam version — watermark-evaluation panics are typically fixed in releases
Work around by simplifying pipeline topology (avoid exotic triggers/windows) until fixed

Defensive patterns

Strategy: try-catch

Try / catch

if err := j.WaitUntilDone(ctx); err != nil {
  if strings.Contains(err.Error(), "panic in ElementManager.Bundles") {
    // report to Beam with the embedded stacktrace
  }
}

Prevention

Test complex windowing/test-stream pipelines on small data before production
Track Beam releases for watermark-evaluation fixes

When it happens

Trigger: Any panic inside ElementManager.Bundles bundle-generation/watermark logic — e.g. nil map access or index-out-of-range while processing stage events or test-stream elements.

Common situations: Pipelines with unusual topologies or TestStream inputs hitting unhandled edge cases in prism's watermark tracking; runner-side bug rather than user misconfiguration.

Understand the failure class

Background: "This is a bug, please report it": internal invariant violations, unreachable panics, and SNH errors explained — this error's family across 47 libraries.

Related errors

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