watermark estimation method

Error message

watermark estimation method %v is defined on non-splittable DoFn. Watermarkestimation is only valid on splittable DoFns

What it means

Watermark estimation (a CreateWatermarkEstimator method) is only valid on splittable DoFns. Beam's validateIsWatermarkEstimating rejects a DoFn that defines CreateWatermarkEstimator without any SDF methods, since watermark estimation depends on restriction splitting.

Solutions

Remove the CreateWatermarkEstimator method if the DoFn is not splittable.
Convert the DoFn into a proper SDF (CreateInitialRestriction, SplitRestriction, CreateTracker) if watermark estimation is genuinely needed.
Use a separate watermark strategy or fixed watermarks for non-SDF sources.

Example fix

// before
func (fn *plainDoFn) ProcessElement(ctx context.Context, x int, emit func(int)) {}
func (fn *plainDoFn) CreateWatermarkEstimator() sdf.WatermarkEstimator { ... }
// after: delete CreateWatermarkEstimator, or add SDF methods
//   (CreateInitialRestriction, SplitRestriction, CreateTracker) alongside it

Defensive patterns

Strategy: validation

Validate before calling

func watermarkRequiresSdf(fn interface{}) error {
    t := reflect.TypeOf(fn)
    _, hasWME := t.MethodByName("CreateWatermarkEstimator")
    _, hasRestr := t.MethodByName("CreateInitialRestriction")
    if hasWME && !hasRestr {
        return errors.New("CreateWatermarkEstimator defined on non-splittable DoFn")
    }
    return nil
}

Type guard

func isSplittable(fn interface{}) bool {
    t := reflect.TypeOf(fn)
    _, a := t.MethodByName("CreateInitialRestriction")
    _, b := t.MethodByName("SplitRestriction")
    _, c := t.MethodByName("CreateTracker")
    return a && b && c
}

Prevention

Only add watermark estimation methods to DoFns that already implement the SDF method set.
Validate DoFns in unit tests with beam.TryCreateDoFn.

When it happens

Trigger: Adding sdf.CreateWatermarkEstimator to a plain (non-splittable) DoFn and submitting the pipeline.

Common situations: Copying watermark-estimation code from an SDF example into a simple DoFn; misunderstanding that watermark estimation requires SDF semantics.

Understand the failure class

Background: UnsupportedOperationException and "is not supported" errors: when a library deliberately refuses a call — this error's family across 30 libraries.

Related errors

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

Appendix: source

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

		if got, want := method.Param[i+startIndex].T, processFn.Param[pos+i].T; got != want {
			err := errors.Errorf("mismatched element type in method %v, param %v. got: %v, want: %v",
				name, idx, got, want)
			return errors.SetTopLevelMsgf(err, "Mismatched element type in method %v, "+
				"parameter at index %v. Got: %v, Want: %v (from method %v). "+
				"Ensure that element parameters in SDF methods have consistent types with element parameters in %v.",
				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 {

View on GitHub (pinned to 12126d8942)