apache/beam · error

unmatched CoGBK boundedness

Error message

unmatched CoGBK boundedness: %v, want %v

What it means

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.

Solutions

1. Ensure all PCollections passed to CoGroupByKey come from sources with the same boundedness (all bounded or all unbounded).
2. If combining batch and streaming data, materialize one side (e.g., via a side input or a bounded connector) before the CoGBK.
3. Check pipeline mode configuration (batch vs streaming runner options) so boundedness is consistent.
4. If this arises inside custom graph-building code, pass the correct `bounded` value matching the node's Bounded() result.

Example fix

// before
coGBK := beam.CoGroupByKey(s, boundedColl, unboundedColl) // panics/fails at graph time
// after
// use a bounded side input instead of an unbounded PCollection in the CoGBK
coGBK := beam.CoGroupByKey(s, boundedColl, beam.Create(s, loadedSnapshot...))

// Go: check boundedness agreement before CoGroupByKey
func boundednessOK(pcs ...beam.PCollection) bool {
    for _, pc := range pcs[1:] {
        if pc.Bounded() != pcs[0].Bounded() { return false }
    }
    return true
}
if !boundednessOK(pc1, pc2) { /* convert one side to bounded or restructure pipeline */ }

When it happens

Trigger: Calling TryCoGroupByKey/NewCoGBK with multiple PCollections where one is bounded (from BoundedSource/batch pipeline) and another is unbounded (streaming source), so `bounded != n.Bounded()`.

Common situations: 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.

Understand the failure class

Background: "Invalid state transition" errors: "status must be X, actually Y", "already rejected/charging/uninstalled", "cannot ... while running" — what they mean when a library rejects your call — this error's family across 31 libraries.

Related errors

• Combine requires CoGBK type
• Flatten needs at least 2 input, got
• mismatched Flatten input types
• mismatched Flatten window types
• cannot make a keyed iterable for an unkeyed side input

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

Appendix: source

Thrown at sdks/go/pkg/beam/core/graph/edge.go:231

	c := ns[0].Coder.Components[0]
	w := inputWindow(ns)
	bounded := inputBounded(ns)
	comp := []typex.FullType{c.T, ns[0].Type().Components()[1]}

	for i := 1; i < len(ns); i++ {
		n := ns[i]
		if !typex.IsKV(n.Type()) {
			return nil, addContext(errors.Errorf("input type must be KV: %v", n), s)
		}
		if !n.Coder.Components[0].Equals(c) {
			return nil, addContext(errors.Errorf("key coder for %v is %v, want %v", n, n.Coder.Components[0], c), s)
		}
		if !w.Equals(n.WindowingStrategy()) {
			return nil, addContext(errors.Errorf("mismatched CoGBK windowing strategies: %v, want %v", n.WindowingStrategy(), w), s)
		}
		if bounded != n.Bounded() {
			return nil, addContext(errors.Errorf("unmatched CoGBK boundedness: %v, want %v", n.Bounded(), bounded), s)
		}

		comp = append(comp, n.Type().Components()[1])
	}

	t := typex.NewCoGBK(comp...)
	out := g.NewNode(t, w, bounded)

	// (2) Add CoGBK edge

	edge := g.NewEdge(s)
	edge.Op = CoGBK
	for i := 0; i < len(ns); i++ {
		edge.Input = append(edge.Input, &Inbound{Kind: Main, From: ns[i], Type: ns[i].Type()})
	}
	edge.Output = []*Outbound{{To: out, Type: t}}
	return edge, nil
}

View on GitHub (pinned to 12126d8942)