not all required stateful watermark estimation methods are…

Error message

not all required stateful watermark estimation methods are present, but CreateWatermarkEstimator takes in a state parameter. Missing methods: %v

What it means

A stateful watermark estimator requires a full set of SDF methods: InitialWatermarkEstimatorState, CreateWatermarkEstimator (taking state), RunWatermarkEstimator, and WatermarkEstimatorState. If CreateWatermarkEstimator takes a state parameter but any of the other required methods is missing, Beam fails validation with this error listing the missing method names.

Solutions

Add every required stateful method named in the error's 'Missing methods' list.
Fix method name typos so Beam's reflection finds them (e.g. WatermarkEstimatorState spelled exactly).
Keep the whole method set (CreateInitialRestriction, CreateTracker, all watermark methods) consistent with the stateful pattern.
If stateless estimation is intended, remove the state parameter so CreateWatermarkEstimator takes only the restriction.

Example fix

// before: only CreateWatermarkEstimator(rest R, st myState) exists
// after: add the missing companion methods
func (f *fn) InitialWatermarkEstimatorState(rt typex.EventTime, rest MyRestriction) myState { return myState{} }
func (f *fn) WatermarkEstimatorState(we sdf.WatermarkEstimator) myState {
    return we.(*myWatermarkEstimator).state
}
func (f *fn) RunWatermarkEstimator(ctx context.Context, we sdf.WatermarkEstimator, bc beam.BoundedSourceBundle, emt *sdf.EmitWatermark) error { ... }

Defensive patterns

Strategy: validation

Validate before calling

// Ensure all stateful watermark methods exist at init time
func init() {
    for _, m := range []string{"InitialWatermarkEstimatorState", "CreateWatermarkEstimator", "RunWatermarkEstimator", "WatermarkEstimatorState"} {
        if !hasMethod(&fn{}, m) { panic("missing SDF method: " + m) }
    }
}

Prevention

When going stateful, add all four watermark methods together in one commit.
Spell method names exactly as the sdf package expects.
Copy the complete method set from a stateful SDF example in the Beam repo.

When it happens

Trigger: Defining CreateWatermarkEstimator(rest R, state State) (the stateful two-parameter variant) on an SDF while omitting one of: InitialWatermarkEstimatorState, WatermarkEstimatorState, RunWatermarkEstimator; or defining them with names Beam's reflection doesn't match.

Common situations: Upgrading an SDF from stateless to stateful estimation by adding a state parameter but forgetting the state-management methods; a typo in a method name; deleting one method during refactoring.

Related errors

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

Appendix: source

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

		return errors.SetTopLevelMsgf(err, "Mismatched output type in method %v, "+
			"return value at index %v. Got: %v, Want: %v (from method %v).",
			watermarkEstimatorStateName, 0, method.Ret[0].T, processFn.Param[pos].T, processElementName)
	}

	return nil
}

func validateStatefulWatermarkSig(fn *Fn, numMainIn int) error {
	// Store missing method names so we can output them to the user if validation fails.
	var missing []string
	for _, name := range watermarkEstimationNames {
		_, ok := fn.methods[name]
		if !ok {
			missing = append(missing, name)
		}
	}
	if len(missing) > 0 {
		err := errors.Errorf("not all required stateful watermark estimation methods are present, "+
			"but CreateWatermarkEstimator takes in a state parameter. Missing methods: %v", missing)
		return err
	}

	restT := fn.methods[createInitialRestrictionName].Ret[0].T
	watermarkStateT := fn.methods[createWatermarkEstimatorName].Param[0].T
	watermarkEstimatorT := fn.methods[createWatermarkEstimatorName].Ret[0].T

	// If number of main inputs is ambiguous, we check for consistency against
	// CreateInitialRestriction.
	if numMainIn == int(MainUnknown) {
		initialRestFn := fn.methods[createInitialRestrictionName]
		paramNum := len(initialRestFn.Params(funcx.FnValue))

		switch paramNum {
		case int(MainSingle), int(MainKv):
			numMainIn = paramNum
		}

View on GitHub (pinned to 12126d8942)