apache/beam · error

Invalid emitter. Emitter

Error message

Invalid emitter. Emitter %v must implement ReusableTimestampObservingWatermarkEmitter interface because it is used in a ParDo with a timestamp observing estimator. If you are not using a custom emitter, you may need to regenerate your shims with the code generator.

What it means

When a ParDo uses a timestamp-observing estimator (attached watermark estimator), every emitter must implement ReusableTimestampObservingWatermarkEmitter so the estimator can be attached and values cached. This error is thrown in initSideInput when an emitter type lacks that interface, typically because generated shim code is stale.

Solutions

1. Regenerate the shims: run the beam code generator (go generate ./... on sdks/go/pkg/beam/runners/v2 or the worker's codegen) so emitters implement the new interface.
2. Update the Go SDK to the version matching your generated code (or vice versa).
3. If using a custom emitter, make it implement ReusableTimestampObservingWatermarkEmitter (AttachEstimator and Value methods).

Example fix

// before
type myEmitter struct{ ... }
func (e *myEmitter) Emit(v T) { ... }
// after
type myEmitter struct{ ... }
func (e *myEmitter) Emit(v T) { ... }
func (e *myEmitter) AttachEstimator(est *WatermarkEstimator) { e.est = est }
func (e *myEmitter) Value() typex.EventTime { return e.est.CurrentWatermark() }

for _, emit := range emitters {
    if _, ok := emit.(exec.ReusableTimestampObservingWatermarkEmitter); !ok {
        return fmt.Errorf("emitter %T does not implement ReusableTimestampObservingWatermarkEmitter; regenerate shims", emit)
    }
}

func isReusableWatermarkEmitter(e interface{}) bool {
    _, ok := e.(exec.ReusableTimestampObservingWatermarkEmitter)
    return ok
}

if err := plan.Execute(ctx, id, mgr); err != nil {
    if strings.Contains(err.Error(), "ReusableTimestampObservingWatermarkEmitter") {
        return fmt.Errorf("regenerate shims with the beam code generator: %w", err)
    }
}

When it happens

Trigger: Running a pipeline whose DoFn shims were generated by an older beam version, before the timestamp-observing emitter interface existed; using a custom emitter type that does not implement ReusableTimestampObservingWatermarkEmitter while an estimator is attached.

Common situations: Upgrading the Apache Beam Go SDK without regenerating code-shims (via the code generator); hand-written custom emitters in SDF/estimator pipelines.

Understand the failure class

Background: "is deprecated and will be removed" — deprecation warnings for old API names, keywords, and options, and how to migrate before the removal release — this error's family across 29 libraries.

Related errors

• CreateTracker has unexpected number of return values
• AfterProcessingTime trigger set without a delay or…
• array len mismatch. decoding
• At least one subtrigger required for composite triggers.
• attempted to add namespace to missing coder id

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

Appendix: source

Thrown at sdks/go/pkg/beam/core/runtime/exec/pardo.go:311

		n.cache = &cacheElm{
			key:   w,
			extra: make([]any, sideCount+emitCount),
		}
		attachEstimator := false
		if n.we != nil {
			// var ok bool
			if _, ok := n.we.(sdf.TimestampObservingEstimator); ok {
				attachEstimator = true
			}
		}
		for i, emit := range n.emitters {
			if attachEstimator {
				if weEmit, ok := emit.(ReusableTimestampObservingWatermarkEmitter); ok {
					weEmit.AttachEstimator(&n.we)
					n.cache.extra[i+sideCount] = weEmit.Value()
				} else {
					return errors.Errorf("Invalid emitter. Emitter %v must implement "+
						"ReusableTimestampObservingWatermarkEmitter interface because it is "+
						"used in a ParDo with a timestamp observing estimator. If you are not "+
						"using a custom emitter, you may need to regenerate your shims with the code "+
						"generator.", reflect.TypeOf(emit))
				}
			} else {
				n.cache.extra[i+sideCount] = emit.Value()
			}
		}
	} else if w.Equals(n.cache.key) {
		// Fast path: same window. Just unwind the side inputs.

		for _, s := range n.cache.sideinput {
			if err := s.Init(); err != nil {
				return err
			}
		}
		return nil

View on GitHub (pinned to 12126d8942)