{"record":{"id":"19747b934c57b561","repo":"apache/beam","slug":"unmatched-cogbk-boundedness-v-want-v","errorCode":null,"errorMessage":"unmatched CoGBK boundedness: %v, want %v","messagePattern":"unmatched CoGBK boundedness: (.+?), want (.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/core/graph/edge.go","lineNumber":231,"sourceCode":"\n\tc := ns[0].Coder.Components[0]\n\tw := inputWindow(ns)\n\tbounded := inputBounded(ns)\n\tcomp := []typex.FullType{c.T, ns[0].Type().Components()[1]}\n\n\tfor i := 1; i < len(ns); i++ {\n\t\tn := ns[i]\n\t\tif !typex.IsKV(n.Type()) {\n\t\t\treturn nil, addContext(errors.Errorf(\"input type must be KV: %v\", n), s)\n\t\t}\n\t\tif !n.Coder.Components[0].Equals(c) {\n\t\t\treturn nil, addContext(errors.Errorf(\"key coder for %v is %v, want %v\", n, n.Coder.Components[0], c), s)\n\t\t}\n\t\tif !w.Equals(n.WindowingStrategy()) {\n\t\t\treturn nil, addContext(errors.Errorf(\"mismatched CoGBK windowing strategies: %v, want %v\", n.WindowingStrategy(), w), s)\n\t\t}\n\t\tif bounded != n.Bounded() {\n\t\t\treturn nil, addContext(errors.Errorf(\"unmatched CoGBK boundedness: %v, want %v\", n.Bounded(), bounded), s)\n\t\t}\n\n\t\tcomp = append(comp, n.Type().Components()[1])\n\t}\n\n\tt := typex.NewCoGBK(comp...)\n\tout := g.NewNode(t, w, bounded)\n\n\t// (2) Add CoGBK edge\n\n\tedge := g.NewEdge(s)\n\tedge.Op = CoGBK\n\tfor i := 0; i < len(ns); i++ {\n\t\tedge.Input = append(edge.Input, &Inbound{Kind: Main, From: ns[i], Type: ns[i].Type()})\n\t}\n\tedge.Output = []*Outbound{{To: out, Type: t}}\n\treturn edge, nil\n}","sourceCodeStart":213,"sourceCodeEnd":249,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/core/graph/edge.go#L213-L249","documentation":"NewCoGBK in the Apache Beam Go SDK validates that every input node has a matching boundedness flag when building (or reusing) a CoGBK node. If the `bounded` argument passed by the caller does not match the boundedness of the existing/derived CoGBK node `n`, graph construction fails with this error wrapped in a scope context. It is an internal consistency check ensuring all inputs to a grouped stream agree on bounded vs unbounded semantics.","triggerScenarios":"Calling TryCoGroupByKey/NewCoGBK with multiple PCollections where one is bounded (from BoundedSource/batch pipeline) and another is unbounded (streaming source), so `bounded != n.Bounded()`.","commonSituations":"Mixing batch (bounded) and streaming (unbounded) sources in a CoGroupByKey; applying a windowing/re-grouping helper that infers boundedness from one input while another differs; runner-generated graphs after a source transition.","solutions":["Ensure all PCollections passed to CoGroupByKey come from sources with the same boundedness (all bounded or all unbounded).","If combining batch and streaming data, materialize one side (e.g., via a side input or a bounded connector) before the CoGBK.","Check pipeline mode configuration (batch vs streaming runner options) so boundedness is consistent.","If this arises inside custom graph-building code, pass the correct `bounded` value matching the node's Bounded() result."],"exampleFix":"// before\ncoGBK := beam.CoGroupByKey(s, boundedColl, unboundedColl) // panics/fails at graph time\n// after\n// use a bounded side input instead of an unbounded PCollection in the CoGBK\ncoGBK := beam.CoGroupByKey(s, boundedColl, beam.Create(s, loadedSnapshot...))","handlingStrategy":"validation","validationCode":"// Go: check boundedness agreement before CoGroupByKey\nfunc boundednessOK(pcs ...beam.PCollection) bool {\n    for _, pc := range pcs[1:] {\n        if pc.Bounded() != pcs[0].Bounded() { return false }\n    }\n    return true\n}\nif !boundednessOK(pc1, pc2) { /* convert one side to bounded or restructure pipeline */ }","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Never mix batch and streaming sources in a single CoGroupByKey.","Materialize streaming sides as bounded side inputs when combining with batch data.","Keep pipeline mode (batch/streaming) uniform across all inputs."],"tags":["beam-go","cogroupbykey","boundedness","graph-construction"],"backgroundTag":"invalid-state-transition","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}