{"record":{"id":"3ee693740bbb7e5b","repo":"apache/beam","slug":"failed-to-merge-windows-got-v","errorCode":null,"errorMessage":"failed to merge windows, got: %v","messagePattern":"failed to merge windows, got: (.+?)","errorType":"error_code","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/runners/direct/gbk.go","lineNumber":105,"sourceCode":"\t}\n\tkey := buf.String()\n\tg, ok := m[key]\n\tif !ok {\n\t\tg = &group{\n\t\t\tkey:    exec.FullValue{Elm: elm.Elm, Timestamp: elm.Timestamp, Windows: ws},\n\t\t\tvalues: make([][]exec.FullValue, len(n.Edge.Input)),\n\t\t}\n\t\tm[key] = g\n\t}\n\treturn g, nil\n}\n\nfunc (n *CoGBK) FinishBundle(ctx context.Context) error {\n\twinKind := n.Edge.Input[0].From.WindowingStrategy().Fn.Kind\n\tif winKind == window.Sessions {\n\t\tmergeMap, mergeErr := n.mergeWindows()\n\t\tif mergeErr != nil {\n\t\t\treturn errors.Errorf(\"failed to merge windows, got: %v\", mergeErr)\n\t\t}\n\t\tif reprocessErr := n.reprocessByWindow(mergeMap); reprocessErr != nil {\n\t\t\treturn errors.Errorf(\"failed to reprocess with merged windows, got :%v\", reprocessErr)\n\t\t}\n\t}\n\tfor key, g := range n.m {\n\t\tvalues := make([]exec.ReStream, len(g.values))\n\t\tfor i, list := range g.values {\n\t\t\tvalues[i] = &exec.FixedReStream{Buf: list}\n\t\t}\n\t\tif err := n.Out.ProcessElement(ctx, &g.key, values...); err != nil {\n\t\t\treturn err\n\t\t}\n\t\tdelete(n.m, key)\n\t}\n\treturn n.Out.FinishBundle(ctx)\n}\n","sourceCodeStart":87,"sourceCodeEnd":123,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/runners/direct/gbk.go#L87-L123","documentation":"At FinishBundle, if the CoGBK input uses session windowing, the node merges accumulated windows via mergeWindows. Any error from that merge (inconsistent window state, failures hashing/comparing session windows) is surfaced as 'failed to merge windows, got: %v' and fails the bundle.","triggerScenarios":"FinishBundle on a CoGBK node whose input windowing strategy Kind is window.Sessions and where n.mergeWindows() returns an error — e.g. session windows that cannot be consistently merged from the buffered per-key data.","commonSituations":"Session-windowed CoGBK pipelines with many overlapping windows; pipelines combining sessions with late data; bugs or edge cases in custom session WindowFns; large in-memory session state in the direct runner.","solutions":["Read the embedded mergeErr to see which windows/state failed to merge.","Validate your session gap duration and that the WindowFn is a standard window.Sessions strategy.","Reduce overlapping/unbounded session accumulation or switch the runner (Dataflow/Flink handle merging natively).","Upgrade the Beam SDK — session-merging bugs in the direct runner's CoGBK have been fixed over releases."],"exampleFix":"// before\nwindow.NewSessions(0 * time.Second) // degenerate gap causes merge issues\n// after\nwindow.NewSessions(5 * time.Minute)","handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"err := plan.Execute(ctx, \"\", exec.DataContext{})\nif err != nil && strings.Contains(err.Error(), \"failed to merge windows\") {\n    log.Printf(\"session merge failed; check session WindowFn/gap: %v\", err)\n}","preventionTips":["Use window.NewSessions with a positive gap duration.","Prefer production runners for session-windowed CoGBK workloads.","Keep the Beam SDK up to date for windowing fixes."],"tags":["go","beam","direct-runner","windowing","sessions"],"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-14T21:17:11.552Z"}