stateful watermark estimation method

Error message

stateful watermark estimation method %v is present, but CreateWatermarkEstimator doesn't take in a state parameter.

What it means

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.

Solutions

Add the state parameter to CreateWatermarkEstimator matching the type returned by InitialWatermarkEstimatorState.
Or remove InitialWatermarkEstimatorState/WatermarkEstimatorState if stateful estimation is not intended.

Example fix

// before
func (fn *f) InitialWatermarkEstimatorState(rt *sdf.LockRTracker) StateT {...}
func (fn *f) CreateWatermarkEstimator() *myEstimator {...}
// after
func (fn *f) CreateWatermarkEstimator(state StateT) *myEstimator {...}

Defensive patterns

Strategy: validation

Validate before calling

func statefulWmConsistent(fn interface{}) error {
    t := reflect.TypeOf(fn)
    if _, has := t.MethodByName("InitialWatermarkEstimatorState"); !has { return nil }
    m, hasCreator := t.MethodByName("CreateWatermarkEstimator")
    if !hasCreator || m.Type.NumIn() != 2 {
        return errors.New("InitialWatermarkEstimatorState present but CreateWatermarkEstimator lacks a state parameter")
    }
    return nil
}

Prevention

Define the stateful estimator trio (InitialWatermarkEstimatorState, WatermarkEstimatorState, state-taking CreateWatermarkEstimator) as one unit.
Match the state parameter type exactly to InitialWatermarkEstimatorState's return type.

When it happens

Trigger: Defining InitialWatermarkEstimatorState() on the DoFn while CreateWatermarkEstimator() has zero parameters.

Common situations: Adding stateful estimator methods incrementally and forgetting the creator's parameter; copying stateful estimator methods from another DoFn but keeping a parameterless CreateWatermarkEstimator.

Related errors

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

Appendix: source

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

	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 {
		err := errors.Errorf("unexpected number of params in method %v. got: %v, want number in range: 0 to 1",
			createWatermarkEstimatorName, len(method.Param))
		return errors.SetTopLevelMsgf(err, "unexpected number of parameters in method %v. "+
			"got: %v, want number in range: 0 to 1. Check that the signature conforms to the expected signature for %v.",
			createWatermarkEstimatorName, len(method.Param), createWatermarkEstimatorName)
	} else if len(method.Param) == 1 {
		err := validateStatefulWatermarkSig(fn, numMainIn)
		if err != nil {
			return err
		}
	} else {
		if _, ok := fn.methods[initialWatermarkEstimatorStateName]; ok {
			err := errors.Errorf("stateful watermark estimation method %v is present, "+
				"but CreateWatermarkEstimator doesn't take in a state parameter.", initialWatermarkEstimatorStateName)
			return err
		}
		if _, ok := fn.methods[watermarkEstimatorStateName]; ok {
			err := errors.Errorf("stateful watermark estimation method %v is present, "+
				"but CreateWatermarkEstimator doesn't take in a state parameter.", watermarkEstimatorStateName)
			return err
		}
	}

	if len(method.Ret) != returnNum {
		err := errors.Errorf("unexpected number of returns in method %v. got: %v, want: %v",
			createWatermarkEstimatorName, len(method.Ret), returnNum)
		return errors.SetTopLevelMsgf(err, "unexpected number of return values in method %v. "+
			"got: %v, want: %v. Check that the signature conforms to the expected signature for %v.",
			createWatermarkEstimatorName, len(method.Ret), returnNum, createWatermarkEstimatorName)
	} else if !method.Ret[0].T.Implements(watermarkEstimatorT) {
		err := errors.Errorf("invalid output type in method %v, return %v: %v does not implement sdf.WatermarkEstimator",

View on GitHub (pinned to 12126d8942)