{"record":{"id":"8c1e6c15cafcf107","repo":"apache/beam","slug":"failed-to-reprocess-with-merged-windows-got-v","errorCode":null,"errorMessage":"failed to reprocess with merged windows, got :%v","messagePattern":"failed to reprocess with merged windows, got :(.+?)","errorType":"error_code","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/runners/direct/gbk.go","lineNumber":108,"sourceCode":"\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\nfunc (n *CoGBK) mergeWindows() (map[typex.Window]int, error) {\n\tsort.Slice(n.wins, func(i int, j int) bool {\n\t\treturn n.wins[i].MaxTimestamp() < n.wins[j].MaxTimestamp()","sourceCodeStart":90,"sourceCodeEnd":126,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/runners/direct/gbk.go#L90-L126","documentation":"After successfully merging session windows in FinishBundle, the CoGBK node reprocesses its buffered values under the merged windows via reprocessByWindow. If that reprocessing step fails, the bundle is failed with 'failed to reprocess with merged windows, got :%v'. It indicates the merged-window replay could not complete for the buffered data.","triggerScenarios":"FinishBundle on a session-windowed CoGBK node where mergeWindows succeeded but reprocessByWindow(mergeMap) returns an error — e.g. a buffered key/window missing from the merge map or an internal stream-construction failure during replay.","commonSituations":"Session windows with many merged groups; pipelines where elements arrive in windows not represented in the merge result; SDK bugs in the direct runner's reprocessing logic; very large buffered bundles exhausting assumptions of the replay path.","solutions":["Read the embedded reprocessErr to pinpoint which key/window failed replay.","Ensure all inputs to the CoGBK share the same windowing strategy and coder; mixed windows confuse the merge map.","Switch to a production runner (Dataflow/Flink) for session-windowed CoGBK workloads — the direct runner is for testing only.","Update the Beam Go SDK to the latest version to pick up fixes to the CoGBK session-reprocessing path."],"exampleFix":"// before\nw1 := beam.WindowInto(s, window.NewSessions(time.Minute), colA)\nw2 := beam.WindowInto(s, window.NewFixedWindows(time.Hour), colB) // mismatched strategies into same CoGBK\n// after\nw1 := beam.WindowInto(s, window.NewSessions(time.Minute), colA)\nw2 := beam.WindowInto(s, window.NewSessions(time.Minute), colB)","handlingStrategy":"try-catch","validationCode":"// ensure all CoGBK inputs share the same windowing strategy\nfor _, col := range inputs {\n    if col.WindowingStrategy().Fn.Kind != window.Sessions {\n        return fmt.Errorf(\"CoGBK input must use session windows\")\n    }\n}","typeGuard":null,"tryCatchPattern":"err := plan.Execute(ctx, \"\", exec.DataContext{})\nif err != nil && strings.Contains(err.Error(), \"failed to reprocess with merged windows\") {\n    log.Printf(\"session replay failed; align windowing across CoGBK inputs: %v\", err)\n}","preventionTips":["Apply identical session windowing to every CoGBK input.","Avoid mixing window kinds feeding one CoGBK.","Validate session pipelines on the direct runner only with small data; use Dataflow/Flink in production."],"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"}