apache/beam · error

Method has an sdf.WatermarkEstimator parameter at index …

Error message

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.

What it means

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.

Solutions

1. Add the required CreateWatermarkEstimator method (and make the DoFn a splittable DoFn, since that is required).
2. Remove the sdf.WatermarkEstimator parameter from ProcessElement if watermark estimation is not needed.
3. Check the reported parameter index and remove other stray WatermarkEstimator parameters accordingly.

Example fix

// before
func (fn *f) ProcessElement(ctx, we sdf.WatermarkEstimator, emit func(int)) {}
// after: add
func (fn *f) CreateWatermarkEstimator() sdf.WatermarkEstimator { return sdf.NewTimestampedWatermarkEstimator(...) }
// (and make the DoFn splittable), or drop the parameter

func wmParamNeedsCreator(fn interface{}) error {
    t := reflect.TypeOf(fn)
    pe, ok := t.MethodByName("ProcessElement")
    if !ok { return nil }
    we := reflect.TypeOf((*sdf.WatermarkEstimator)(nil)).Elem()
    for i := 1; i < pe.Type.NumIn(); i++ {
        if pe.Type.In(i) == we {
            if _, has := t.MethodByName("CreateWatermarkEstimator"); !has {
                return fmt.Errorf("ProcessElement param %d is sdf.WatermarkEstimator but CreateWatermarkEstimator is missing", i-1)
            }
        }
    }
    return nil
}

When it happens

Trigger: Declaring ProcessElement(ctx, ws sdf.WatermarkEstimator, ...) on a DoFn that lacks a CreateWatermarkEstimator method; the error reports the parameter position 'pos'.

Common situations: Removing/renaming CreateWatermarkEstimator while keeping its parameter in ProcessElement; copying a signature from a watermark-estimating example.

Related errors

• watermark estimation method
• CreateInitialRestriction has unexpected number of return…
• CreateTracker has unexpected number of parameters
• CreateTracker has unexpected number of return values
• Invalid output type in method

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/d2123afffbe259b6. Report an issue: GitHub.

Appendix: source

Thrown at sdks/go/pkg/beam/core/graph/fn.go:1164

				name, idx, got, want, processElementName, processElementName)
		}
	}
	return nil
}

// validateIsWatermarkEstimating returns true if watermark estimator methods are present on the DoFn, returns
// false if they aren't, and returns an error if they are present but the function isn't an sdf and thus doesn't
// support watermark estimation
func validateIsWatermarkEstimating(fn *Fn, isSdf bool) (bool, error) {
	_, isWatermarkEstimating := fn.methods[createWatermarkEstimatorName]
	if !isSdf && isWatermarkEstimating {
		return false, errors.Errorf("watermark estimation method %v is defined on non-splittable DoFn. Watermark"+
			"estimation is only valid on splittable DoFns", createWatermarkEstimatorName)
	}

	processFn := fn.methods[processElementName]
	if pos, ok := processFn.WatermarkEstimator(); ok && !isWatermarkEstimating {
		err := errors.Errorf("method %v has sdf.WatermarkEstimator as param %v, expected none",
			processElementName, pos)
		return false, errors.SetTopLevelMsgf(err, "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.",
			processElementName, pos, processElementName)
	}

	return isWatermarkEstimating, nil
}

// validateWatermarkSig validates that all watermark related functions are valid
func validateWatermarkSig(fn *Fn, numMainIn int) error {
	returnNum := 1 // TODO(BEAM-3301): Enable optional error params in SDF methods.

	watermarkEstimatorT := reflect.TypeOf((*sdf.WatermarkEstimator)(nil)).Elem()
	method := fn.methods[createWatermarkEstimatorName]

	if len(method.Param) > 1 {

View on GitHub (pinned to 12126d8942)