{"record":{"id":"5abbd530afeb9dc0","repo":"apache/beam","slug":"invalid-status-for-combine-merge-v-v","errorCode":null,"errorMessage":"invalid status for combine merge %v: %v","messagePattern":"invalid status for combine merge (.+?): (.+?)","errorType":"validation","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/core/runtime/exec/combine.go","lineNumber":577,"sourceCode":"\t}\n\tc.cache = nil\n\treturn nil\n}\n\n// MergeAccumulators is an executor for merging accumulators from a lifted combine.\ntype MergeAccumulators struct {\n\t*Combine\n}\n\nfunc (n *MergeAccumulators) String() string {\n\treturn fmt.Sprintf(\"MergeAccumulators[%v] Keyed:%v Out:%v\", path.Base(n.Fn.Name()), n.UsesKey, n.Out.ID())\n}\n\n// ProcessElement accepts a stream of accumulator values with the same key and\n// runs the MergeAccumulatorsFn over them repeatedly.\nfunc (n *MergeAccumulators) ProcessElement(ctx context.Context, value *FullValue, values ...ReStream) error {\n\tif n.status != Active {\n\t\treturn errors.Errorf(\"invalid status for combine merge %v: %v\", n.UID, n.status)\n\t}\n\tn.Combine.states.Set(n.Combine.ctx, metrics.ProcessBundle)\n\ta, err := n.newAccum(n.Combine.ctx, value.Elm)\n\tif err != nil {\n\t\treturn n.fail(err)\n\t}\n\tfirst := true\n\n\tstream, err := values[0].Open()\n\tif err != nil {\n\t\treturn n.fail(err)\n\t}\n\tdefer stream.Close()\n\tfor {\n\t\tv, err := stream.Read()\n\t\tif err != nil {\n\t\t\tif err == io.EOF {\n\t\t\t\tbreak","sourceCodeStart":559,"sourceCodeEnd":595,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/core/runtime/exec/combine.go#L559-L595","documentation":"MergeAccumulators.ProcessElement (the GBK merge side of a combine) requires the node to be Active within a started bundle. Elements arriving while the node is in any other status indicate an out-of-order lifecycle invocation.","triggerScenarios":"Calling MergeAccumulators.ProcessElement before StartBundle or after FinishBundle.","commonSituations":"Tests feeding accumulator streams directly to the merge node; runners whose bundle-start step failed silently but continued to emit data.","solutions":["Ensure StartBundle was called successfully before processing accumulator elements.","Check for swallowed errors from earlier lifecycle calls (a failed StartBundle leaves status Up).","Rebuild the plan node if its status is stale."],"exampleFix":null,"handlingStrategy":"validation","validationCode":null,"typeGuard":null,"tryCatchPattern":"if err := merge.ProcessElement(ctx, fv, values); err != nil {\n\tif strings.Contains(err.Error(), \"invalid status for combine merge\") {\n\t\t// ensure the merge node was started (Up + StartBundle)\n\t}\n\treturn err\n}","preventionTips":["Verify StartBundle succeeded before streaming accumulators.","Fail fast on earlier lifecycle errors instead of continuing to process.","Use standard plan construction in tests."],"tags":["go","beam","lifecycle","state-machine","combine"],"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-14T16:17:12.679Z"}