{"record":{"id":"14502e9980dcad48","repo":"apache/beam","slug":"watermark-estimation-method-v-is-defined-on-non-splittable","errorCode":null,"errorMessage":"watermark estimation method %v is defined on non-splittable DoFn. Watermarkestimation is only valid on splittable DoFns","messagePattern":"watermark estimation method (.+?) is defined on non-splittable DoFn\\. Watermarkestimation is only valid on splittable DoFns","errorType":"validation","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/core/graph/fn.go","lineNumber":1158,"sourceCode":"\t\tif got, want := method.Param[i+startIndex].T, processFn.Param[pos+i].T; got != want {\n\t\t\terr := errors.Errorf(\"mismatched element type in method %v, param %v. got: %v, want: %v\",\n\t\t\t\tname, idx, got, want)\n\t\t\treturn errors.SetTopLevelMsgf(err, \"Mismatched element type in method %v, \"+\n\t\t\t\t\"parameter at index %v. Got: %v, Want: %v (from method %v). \"+\n\t\t\t\t\"Ensure that element parameters in SDF methods have consistent types with element parameters in %v.\",\n\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 {","sourceCodeStart":1140,"sourceCodeEnd":1176,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/core/graph/fn.go#L1140-L1176","documentation":"Watermark estimation (a CreateWatermarkEstimator method) is only valid on splittable DoFns. Beam's validateIsWatermarkEstimating rejects a DoFn that defines CreateWatermarkEstimator without any SDF methods, since watermark estimation depends on restriction splitting.","triggerScenarios":"Adding sdf.CreateWatermarkEstimator to a plain (non-splittable) DoFn and submitting the pipeline.","commonSituations":"Copying watermark-estimation code from an SDF example into a simple DoFn; misunderstanding that watermark estimation requires SDF semantics.","solutions":["Remove the CreateWatermarkEstimator method if the DoFn is not splittable.","Convert the DoFn into a proper SDF (CreateInitialRestriction, SplitRestriction, CreateTracker) if watermark estimation is genuinely needed.","Use a separate watermark strategy or fixed watermarks for non-SDF sources."],"exampleFix":"// before\nfunc (fn *plainDoFn) ProcessElement(ctx context.Context, x int, emit func(int)) {}\nfunc (fn *plainDoFn) CreateWatermarkEstimator() sdf.WatermarkEstimator { ... }\n// after: delete CreateWatermarkEstimator, or add SDF methods\n//   (CreateInitialRestriction, SplitRestriction, CreateTracker) alongside it","handlingStrategy":"validation","validationCode":"func watermarkRequiresSdf(fn interface{}) error {\n    t := reflect.TypeOf(fn)\n    _, hasWME := t.MethodByName(\"CreateWatermarkEstimator\")\n    _, hasRestr := t.MethodByName(\"CreateInitialRestriction\")\n    if hasWME && !hasRestr {\n        return errors.New(\"CreateWatermarkEstimator defined on non-splittable DoFn\")\n    }\n    return nil\n}","typeGuard":"func isSplittable(fn interface{}) bool {\n    t := reflect.TypeOf(fn)\n    _, a := t.MethodByName(\"CreateInitialRestriction\")\n    _, b := t.MethodByName(\"SplitRestriction\")\n    _, c := t.MethodByName(\"CreateTracker\")\n    return a && b && c\n}","tryCatchPattern":null,"preventionTips":["Only add watermark estimation methods to DoFns that already implement the SDF method set.","Validate DoFns in unit tests with beam.TryCreateDoFn."],"tags":["go","apache-beam","watermark","splittable-dofn"],"backgroundTag":"unsupported-operation","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"}