{"record":{"id":"a79a370c8b7e19ea","repo":"apache/beam","slug":"stateful-watermark-estimation-method-v-is-present-but","errorCode":null,"errorMessage":"stateful watermark estimation method %v is present, but CreateWatermarkEstimator doesn't take in a state parameter.","messagePattern":"stateful watermark estimation method (.+?) is present, but CreateWatermarkEstimator doesn't take in a state parameter\\.","errorType":"validation","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/core/graph/fn.go","lineNumber":1195,"sourceCode":"\treturnNum := 1 // TODO(BEAM-3301): Enable optional error params in SDF methods.\n\n\twatermarkEstimatorT := reflect.TypeOf((*sdf.WatermarkEstimator)(nil)).Elem()\n\tmethod := fn.methods[createWatermarkEstimatorName]\n\n\tif len(method.Param) > 1 {\n\t\terr := errors.Errorf(\"unexpected number of params in method %v. got: %v, want number in range: 0 to 1\",\n\t\t\tcreateWatermarkEstimatorName, len(method.Param))\n\t\treturn errors.SetTopLevelMsgf(err, \"unexpected number of parameters in method %v. \"+\n\t\t\t\"got: %v, want number in range: 0 to 1. Check that the signature conforms to the expected signature for %v.\",\n\t\t\tcreateWatermarkEstimatorName, len(method.Param), createWatermarkEstimatorName)\n\t} else if len(method.Param) == 1 {\n\t\terr := validateStatefulWatermarkSig(fn, numMainIn)\n\t\tif err != nil {\n\t\t\treturn err\n\t\t}\n\t} else {\n\t\tif _, ok := fn.methods[initialWatermarkEstimatorStateName]; ok {\n\t\t\terr := errors.Errorf(\"stateful watermark estimation method %v is present, \"+\n\t\t\t\t\"but CreateWatermarkEstimator doesn't take in a state parameter.\", initialWatermarkEstimatorStateName)\n\t\t\treturn err\n\t\t}\n\t\tif _, ok := fn.methods[watermarkEstimatorStateName]; ok {\n\t\t\terr := errors.Errorf(\"stateful watermark estimation method %v is present, \"+\n\t\t\t\t\"but CreateWatermarkEstimator doesn't take in a state parameter.\", watermarkEstimatorStateName)\n\t\t\treturn err\n\t\t}\n\t}\n\n\tif len(method.Ret) != returnNum {\n\t\terr := errors.Errorf(\"unexpected number of returns in method %v. got: %v, want: %v\",\n\t\t\tcreateWatermarkEstimatorName, len(method.Ret), returnNum)\n\t\treturn errors.SetTopLevelMsgf(err, \"unexpected number of return values in method %v. \"+\n\t\t\t\"got: %v, want: %v. Check that the signature conforms to the expected signature for %v.\",\n\t\t\tcreateWatermarkEstimatorName, len(method.Ret), returnNum, createWatermarkEstimatorName)\n\t} else if !method.Ret[0].T.Implements(watermarkEstimatorT) {\n\t\terr := errors.Errorf(\"invalid output type in method %v, return %v: %v does not implement sdf.WatermarkEstimator\",","sourceCodeStart":1177,"sourceCodeEnd":1213,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/core/graph/fn.go#L1177-L1213","documentation":"Stateful watermark estimation requires that CreateWatermarkEstimator accept the initial state parameter. Beam raises this error when InitialWatermarkEstimatorState is defined on the DoFn but CreateWatermarkEstimator takes no parameters, so the state has no way to reach the estimator.","triggerScenarios":"Defining InitialWatermarkEstimatorState() on the DoFn while CreateWatermarkEstimator() has zero parameters.","commonSituations":"Adding stateful estimator methods incrementally and forgetting the creator's parameter; copying stateful estimator methods from another DoFn but keeping a parameterless CreateWatermarkEstimator.","solutions":["Add the state parameter to CreateWatermarkEstimator matching the type returned by InitialWatermarkEstimatorState.","Or remove InitialWatermarkEstimatorState/WatermarkEstimatorState if stateful estimation is not intended."],"exampleFix":"// before\nfunc (fn *f) InitialWatermarkEstimatorState(rt *sdf.LockRTracker) StateT {...}\nfunc (fn *f) CreateWatermarkEstimator() *myEstimator {...}\n// after\nfunc (fn *f) CreateWatermarkEstimator(state StateT) *myEstimator {...}","handlingStrategy":"validation","validationCode":"func statefulWmConsistent(fn interface{}) error {\n    t := reflect.TypeOf(fn)\n    if _, has := t.MethodByName(\"InitialWatermarkEstimatorState\"); !has { return nil }\n    m, hasCreator := t.MethodByName(\"CreateWatermarkEstimator\")\n    if !hasCreator || m.Type.NumIn() != 2 {\n        return errors.New(\"InitialWatermarkEstimatorState present but CreateWatermarkEstimator lacks a state parameter\")\n    }\n    return nil\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Define the stateful estimator trio (InitialWatermarkEstimatorState, WatermarkEstimatorState, state-taking CreateWatermarkEstimator) as one unit.","Match the state parameter type exactly to InitialWatermarkEstimatorState's return type."],"tags":["go","apache-beam","watermark","stateful"],"backgroundTag":"method-signature-mismatch","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"}