apache/beam · error

CreateWatermarkEstimator fn

Error message

CreateWatermarkEstimator fn %v has unexpected number of parameters: %v

What it means

When wiring up a CreateWatermarkEstimator function for an SDF, the invoker validates the function's arity. Only 0- or 1-parameter signatures are supported (e.g. func(rest) sdf.WatermarkEstimator); anything else fails at initCallFn with this error. It is raised by newCreateWatermarkEstimatorInvoker during plan construction, before any data flows.

Solutions

1. Change CreateWatermarkEstimator to accept at most one parameter (the restriction).
2. Use a no-parameter form func() sdf.WatermarkEstimator if no restriction data is needed.
3. Move extra values (element, context) into the estimator's state or a struct captured at DoFn construction time.
4. Check the Beam SDF watermark estimator signature docs for the supported forms.

Example fix

// before
func (fn *mySdf) CreateWatermarkEstimator(rt *rangeTracker, el string) sdf.WatermarkEstimator { ... }
// after
func (fn *mySdf) CreateWatermarkEstimator(rt *rangeTracker) sdf.WatermarkEstimator { ... }

t := reflect.TypeOf(fn.CreateWatermarkEstimator)
if t.NumIn() > 1 {
    return fmt.Errorf("CreateWatermarkEstimator must take at most 1 parameter, has %d", t.NumIn())
}

if err := beam.RunWithExporter(...); err != nil {
    if strings.Contains(err.Error(), "CreateWatermarkEstimator fn") {
        // correct the function signature
    }
    return err
}

When it happens

Trigger: Registering a CreateWatermarkEstimatorFn whose signature has 2 or more parameters, e.g. func(rest R, el T) sdf.WatermarkEstimator. The reflection-based invoker only builds call paths for 0 or 1 params and hits the default error case at sdf_invokers.go:376.

Common situations: Developers copying the shape of ProcessElement (which takes element+restriction) into CreateWatermarkEstimator, or adding a context parameter that this hook does not accept.

Understand the failure class

Background: "missing required argument" and "the following required arguments were not provided": what required-argument errors mean and how to fix them — this error's family across 20 libraries.

Related errors

• InitialWatermarkEstimatorState fn
• WatermarkEstimatorState fn
• CreateInitialRestriction has unexpected number of parameters
• CreateTracker has unexpected number of parameters
• RestrictionSize has unexpected number of parameters

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

Appendix: source

Thrown at sdks/go/pkg/beam/core/runtime/exec/sdf_invokers.go:376

			return fnT.Call0x1().(sdf.WatermarkEstimator)
		}
	case reflectx.Func1x1:
		n.call = func(rest any) sdf.WatermarkEstimator {
			return fnT.Call1x1(rest).(sdf.WatermarkEstimator)
		}
	default:
		switch len(n.fn.Param) {
		case 0:
			n.call = func(rest any) sdf.WatermarkEstimator {
				return n.fn.Fn.Call(n.args)[0].(sdf.WatermarkEstimator)
			}
		case 1:
			n.call = func(rest any) sdf.WatermarkEstimator {
				n.args[0] = rest
				return n.fn.Fn.Call(n.args)[0].(sdf.WatermarkEstimator)
			}
		default:
			return errors.Errorf("CreateWatermarkEstimator fn %v has unexpected number of parameters: %v",
				n.fn.Fn.Name(), len(n.fn.Param))
		}
	}
	return nil
}

// Invoke calls CreateWatermarkEstimator given a restriction and returns an sdf.WatermarkEstimator.
func (n *cweInvoker) Invoke(rest any) sdf.WatermarkEstimator {
	return n.call(rest)
}

// Reset zeroes argument entries in the cached slice to allow values to be
// garbage collected after the bundle ends.
func (n *cweInvoker) Reset() {
	for i := range n.args {
		n.args[i] = nil
	}
}

View on GitHub (pinned to 12126d8942)