CreateWatermarkEstimator fn %v has unexpected number of para

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.

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)

Solutions

Change CreateWatermarkEstimator to accept at most one parameter (the restriction).
Use a no-parameter form func() sdf.WatermarkEstimator if no restriction data is needed.
Move extra values (element, context) into the estimator's state or a struct captured at DoFn construction time.
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 { ... }

Defensive patterns

Strategy: validation

Validate before calling

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

Try / catch

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

Prevention

Follow documented CreateWatermarkEstimator signatures (0 or 1 param).
Pass extra data via DoFn struct fields, not parameters.
Run signature-validating unit tests before submitting jobs.

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

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