{"record":{"id":"b8de0c35a28c1190","repo":"apache/beam","slug":"not-all-required-stateful-watermark-estimation-methods-are","errorCode":null,"errorMessage":"not all required stateful watermark estimation methods are present, but CreateWatermarkEstimator takes in a state parameter. Missing methods: %v","messagePattern":"not all required stateful watermark estimation methods are present, but CreateWatermarkEstimator takes in a state parameter\\. Missing methods: (.+?)","errorType":"validation","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/core/graph/fn.go","lineNumber":1243,"sourceCode":"\t\treturn errors.SetTopLevelMsgf(err, \"Mismatched output type in method %v, \"+\n\t\t\t\"return value at index %v. Got: %v, Want: %v (from method %v).\",\n\t\t\twatermarkEstimatorStateName, 0, method.Ret[0].T, processFn.Param[pos].T, processElementName)\n\t}\n\n\treturn nil\n}\n\nfunc validateStatefulWatermarkSig(fn *Fn, numMainIn int) error {\n\t// Store missing method names so we can output them to the user if validation fails.\n\tvar missing []string\n\tfor _, name := range watermarkEstimationNames {\n\t\t_, ok := fn.methods[name]\n\t\tif !ok {\n\t\t\tmissing = append(missing, name)\n\t\t}\n\t}\n\tif len(missing) > 0 {\n\t\terr := errors.Errorf(\"not all required stateful watermark estimation methods are present, \"+\n\t\t\t\"but CreateWatermarkEstimator takes in a state parameter. Missing methods: %v\", missing)\n\t\treturn err\n\t}\n\n\trestT := fn.methods[createInitialRestrictionName].Ret[0].T\n\twatermarkStateT := fn.methods[createWatermarkEstimatorName].Param[0].T\n\twatermarkEstimatorT := fn.methods[createWatermarkEstimatorName].Ret[0].T\n\n\t// If number of main inputs is ambiguous, we check for consistency against\n\t// CreateInitialRestriction.\n\tif numMainIn == int(MainUnknown) {\n\t\tinitialRestFn := fn.methods[createInitialRestrictionName]\n\t\tparamNum := len(initialRestFn.Params(funcx.FnValue))\n\n\t\tswitch paramNum {\n\t\tcase int(MainSingle), int(MainKv):\n\t\t\tnumMainIn = paramNum\n\t\t}","sourceCodeStart":1225,"sourceCodeEnd":1261,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/core/graph/fn.go#L1225-L1261","documentation":"A stateful watermark estimator requires a full set of SDF methods: InitialWatermarkEstimatorState, CreateWatermarkEstimator (taking state), RunWatermarkEstimator, and WatermarkEstimatorState. If CreateWatermarkEstimator takes a state parameter but any of the other required methods is missing, Beam fails validation with this error listing the missing method names.","triggerScenarios":"Defining CreateWatermarkEstimator(rest R, state State) (the stateful two-parameter variant) on an SDF while omitting one of: InitialWatermarkEstimatorState, WatermarkEstimatorState, RunWatermarkEstimator; or defining them with names Beam's reflection doesn't match.","commonSituations":"Upgrading an SDF from stateless to stateful estimation by adding a state parameter but forgetting the state-management methods; a typo in a method name; deleting one method during refactoring.","solutions":["Add every required stateful method named in the error's 'Missing methods' list.","Fix method name typos so Beam's reflection finds them (e.g. WatermarkEstimatorState spelled exactly).","Keep the whole method set (CreateInitialRestriction, CreateTracker, all watermark methods) consistent with the stateful pattern.","If stateless estimation is intended, remove the state parameter so CreateWatermarkEstimator takes only the restriction."],"exampleFix":"// before: only CreateWatermarkEstimator(rest R, st myState) exists\n// after: add the missing companion methods\nfunc (f *fn) InitialWatermarkEstimatorState(rt typex.EventTime, rest MyRestriction) myState { return myState{} }\nfunc (f *fn) WatermarkEstimatorState(we sdf.WatermarkEstimator) myState {\n    return we.(*myWatermarkEstimator).state\n}\nfunc (f *fn) RunWatermarkEstimator(ctx context.Context, we sdf.WatermarkEstimator, bc beam.BoundedSourceBundle, emt *sdf.EmitWatermark) error { ... }","handlingStrategy":"validation","validationCode":"// Ensure all stateful watermark methods exist at init time\nfunc init() {\n    for _, m := range []string{\"InitialWatermarkEstimatorState\", \"CreateWatermarkEstimator\", \"RunWatermarkEstimator\", \"WatermarkEstimatorState\"} {\n        if !hasMethod(&fn{}, m) { panic(\"missing SDF method: \" + m) }\n    }\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["When going stateful, add all four watermark methods together in one commit.","Spell method names exactly as the sdf package expects.","Copy the complete method set from a stateful SDF example in the Beam repo."],"tags":["go","apache-beam","sdf","watermark-estimator","missing-method"],"backgroundTag":"method-not-implemented","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"}