{"record":{"id":"4cacd1abb6d58108","repo":"apache/beam","slug":"panic-in-elementmanager-bundles-watermark-evaluation","errorCode":null,"errorMessage":"panic in ElementManager.Bundles watermark evaluation goroutine: %v\n%v","messagePattern":"panic in ElementManager\\.Bundles watermark evaluation goroutine: (.+?)\n(.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"critical","filePath":"sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go","lineNumber":410,"sourceCode":"\t\tslog.Debug(\"no more pending elements: terminating pipeline\")\n\t\tcancelFn(fmt.Errorf(\"elementManager out of elements, cleaning up\"))\n\t\t// Ensure the watermark evaluation goroutine exits by locking the mutex\n\t\t// before broadcasting, preventing a lost wake-up signal.\n\t\tem.refreshCond.L.Lock()\n\t\tem.refreshCond.Broadcast()\n\t\tem.refreshCond.L.Unlock()\n\t}()\n\t// Watermark evaluation goroutine.\n\tgo func() {\n\t\t// We should defer closing of the channel first, so that when a panic happens,\n\t\t// we will handle the panic and trigger a job failure BEFORE the job is\n\t\t// prematurely marked as done.\n\t\tdefer close(runStageCh)\n\t\tdefer func() {\n\t\t\t// In case of panics in bundle generation, fail and cancel the job.\n\t\t\tif e := recover(); e != nil {\n\t\t\t\tslog.Error(\"panic in ElementManager.Bundles watermark evaluation goroutine\", \"error\", e, \"traceback\", string(debug.Stack()))\n\t\t\t\tupstreamCancelFn(fmt.Errorf(\"panic in ElementManager.Bundles watermark evaluation goroutine: %v\\n%v\", e, string(debug.Stack())))\n\t\t\t}\n\t\t}()\n\n\t\tfor {\n\t\t\tem.refreshCond.L.Lock()\n\t\t\t// Check if processing time has advanced before the wait loop.\n\t\t\temNow := em.processingTimeNow()\n\t\t\tchangedByProcessingTime := em.processTimeEvents.AdvanceTo(emNow)\n\t\t\tem.changedStages.merge(changedByProcessingTime)\n\n\t\t\t// If there are no changed stages, ready processing time events,\n\t\t\t// or injected bundles available, we wait until there are.\n\t\t\tfor len(em.changedStages)+len(changedByProcessingTime)+len(em.injectedBundles) == 0 {\n\t\t\t\t// Check to see if we must exit\n\t\t\t\tselect {\n\t\t\t\tcase <-ctx.Done():\n\t\t\t\t\tem.refreshCond.L.Unlock()\n\t\t\t\t\treturn","sourceCodeStart":392,"sourceCodeEnd":428,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go#L392-L428","documentation":"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.","triggerScenarios":"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.","commonSituations":"Pipelines with unusual topologies or TestStream inputs hitting unhandled edge cases in prism's watermark tracking; runner-side bug rather than user misconfiguration.","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"],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"if err := j.WaitUntilDone(ctx); err != nil {\n  if strings.Contains(err.Error(), \"panic in ElementManager.Bundles\") {\n    // report to Beam with the embedded stacktrace\n  }\n}","preventionTips":["Test complex windowing/test-stream pipelines on small data before production","Track Beam releases for watermark-evaluation fixes"],"tags":["go","beam","prism-runner","panic","watermark"],"backgroundTag":"internal-invariant-violation","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}