{"record":{"id":"d2123afffbe259b6","repo":"apache/beam","slug":"method-v-has-an-sdf-watermarkestimator-parameter-at-index-v","errorCode":null,"errorMessage":"Method %v has an sdf.WatermarkEstimator parameter at index %v, but is not part of a watermark estimating DoFn. sdf.WatermarkEstimator is invalid in %v in non-watermark estimating DoFns.","messagePattern":"Method (.+?) has an sdf\\.WatermarkEstimator parameter at index (.+?), but is not part of a watermark estimating DoFn\\. sdf\\.WatermarkEstimator is invalid in (.+?) in non-watermark estimating DoFns\\.","errorType":"validation","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/core/graph/fn.go","lineNumber":1164,"sourceCode":"\t\t\t\tname, idx, got, want, processElementName, processElementName)\n\t\t}\n\t}\n\treturn nil\n}\n\n// validateIsWatermarkEstimating returns true if watermark estimator methods are present on the DoFn, returns\n// false if they aren't, and returns an error if they are present but the function isn't an sdf and thus doesn't\n// support watermark estimation\nfunc validateIsWatermarkEstimating(fn *Fn, isSdf bool) (bool, error) {\n\t_, isWatermarkEstimating := fn.methods[createWatermarkEstimatorName]\n\tif !isSdf && isWatermarkEstimating {\n\t\treturn false, errors.Errorf(\"watermark estimation method %v is defined on non-splittable DoFn. Watermark\"+\n\t\t\t\"estimation is only valid on splittable DoFns\", createWatermarkEstimatorName)\n\t}\n\n\tprocessFn := fn.methods[processElementName]\n\tif pos, ok := processFn.WatermarkEstimator(); ok && !isWatermarkEstimating {\n\t\terr := errors.Errorf(\"method %v has sdf.WatermarkEstimator as param %v, expected none\",\n\t\t\tprocessElementName, pos)\n\t\treturn false, errors.SetTopLevelMsgf(err, \"Method %v has an sdf.WatermarkEstimator parameter at index %v, \"+\n\t\t\t\"but is not part of a watermark estimating DoFn. sdf.WatermarkEstimator is invalid in %v in \"+\n\t\t\t\"non-watermark estimating DoFns.\",\n\t\t\tprocessElementName, pos, processElementName)\n\t}\n\n\treturn isWatermarkEstimating, nil\n}\n\n// validateWatermarkSig validates that all watermark related functions are valid\nfunc validateWatermarkSig(fn *Fn, numMainIn int) error {\n\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 {","sourceCodeStart":1146,"sourceCodeEnd":1182,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/core/graph/fn.go#L1146-L1182","documentation":"ProcessElement may declare an sdf.WatermarkEstimator parameter only when the DoFn is watermark estimating (i.e. it defines CreateWatermarkEstimator). Beam raises this error when the parameter is present but no watermark estimator method exists, since Beam would have nothing to instantiate it from.","triggerScenarios":"Declaring ProcessElement(ctx, ws sdf.WatermarkEstimator, ...) on a DoFn that lacks a CreateWatermarkEstimator method; the error reports the parameter position 'pos'.","commonSituations":"Removing/renaming CreateWatermarkEstimator while keeping its parameter in ProcessElement; copying a signature from a watermark-estimating example.","solutions":["Add the required CreateWatermarkEstimator method (and make the DoFn a splittable DoFn, since that is required).","Remove the sdf.WatermarkEstimator parameter from ProcessElement if watermark estimation is not needed.","Check the reported parameter index and remove other stray WatermarkEstimator parameters accordingly."],"exampleFix":"// before\nfunc (fn *f) ProcessElement(ctx, we sdf.WatermarkEstimator, emit func(int)) {}\n// after: add\nfunc (fn *f) CreateWatermarkEstimator() sdf.WatermarkEstimator { return sdf.NewTimestampedWatermarkEstimator(...) }\n// (and make the DoFn splittable), or drop the parameter","handlingStrategy":"validation","validationCode":"func wmParamNeedsCreator(fn interface{}) error {\n    t := reflect.TypeOf(fn)\n    pe, ok := t.MethodByName(\"ProcessElement\")\n    if !ok { return nil }\n    we := reflect.TypeOf((*sdf.WatermarkEstimator)(nil)).Elem()\n    for i := 1; i < pe.Type.NumIn(); i++ {\n        if pe.Type.In(i) == we {\n            if _, has := t.MethodByName(\"CreateWatermarkEstimator\"); !has {\n                return fmt.Errorf(\"ProcessElement param %d is sdf.WatermarkEstimator but CreateWatermarkEstimator is missing\", i-1)\n            }\n        }\n    }\n    return nil\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Add or remove CreateWatermarkEstimator and the ProcessElement WatermarkEstimator parameter together.","Grep for sdf.WatermarkEstimator in DoFn files after refactors."],"tags":["go","apache-beam","watermark","splittable-dofn"],"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"}