{"record":{"id":"734c571ea47a1aab","repo":"apache/beam","slug":"invalid-emitter-emitter-v-must-implement","errorCode":null,"errorMessage":"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.","messagePattern":"Invalid emitter\\. Emitter (.+?) 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\\.","errorType":"error_code","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/core/runtime/exec/pardo.go","lineNumber":311,"sourceCode":"\n\t\tn.cache = &cacheElm{\n\t\t\tkey:   w,\n\t\t\textra: make([]any, sideCount+emitCount),\n\t\t}\n\t\tattachEstimator := false\n\t\tif n.we != nil {\n\t\t\t// var ok bool\n\t\t\tif _, ok := n.we.(sdf.TimestampObservingEstimator); ok {\n\t\t\t\tattachEstimator = true\n\t\t\t}\n\t\t}\n\t\tfor i, emit := range n.emitters {\n\t\t\tif attachEstimator {\n\t\t\t\tif weEmit, ok := emit.(ReusableTimestampObservingWatermarkEmitter); ok {\n\t\t\t\t\tweEmit.AttachEstimator(&n.we)\n\t\t\t\t\tn.cache.extra[i+sideCount] = weEmit.Value()\n\t\t\t\t} else {\n\t\t\t\t\treturn errors.Errorf(\"Invalid emitter. Emitter %v must implement \"+\n\t\t\t\t\t\t\"ReusableTimestampObservingWatermarkEmitter interface because it is \"+\n\t\t\t\t\t\t\"used in a ParDo with a timestamp observing estimator. If you are not \"+\n\t\t\t\t\t\t\"using a custom emitter, you may need to regenerate your shims with the code \"+\n\t\t\t\t\t\t\"generator.\", reflect.TypeOf(emit))\n\t\t\t\t}\n\t\t\t} else {\n\t\t\t\tn.cache.extra[i+sideCount] = emit.Value()\n\t\t\t}\n\t\t}\n\t} else if w.Equals(n.cache.key) {\n\t\t// Fast path: same window. Just unwind the side inputs.\n\n\t\tfor _, s := range n.cache.sideinput {\n\t\t\tif err := s.Init(); err != nil {\n\t\t\t\treturn err\n\t\t\t}\n\t\t}\n\t\treturn nil","sourceCodeStart":293,"sourceCodeEnd":329,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/core/runtime/exec/pardo.go#L293-L329","documentation":"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.","triggerScenarios":"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.","commonSituations":"Upgrading the Apache Beam Go SDK without regenerating code-shims (via the code generator); hand-written custom emitters in SDF/estimator pipelines.","solutions":["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.","Update the Go SDK to the version matching your generated code (or vice versa).","If using a custom emitter, make it implement ReusableTimestampObservingWatermarkEmitter (AttachEstimator and Value methods)."],"exampleFix":"// before\ntype myEmitter struct{ ... }\nfunc (e *myEmitter) Emit(v T) { ... }\n// after\ntype myEmitter struct{ ... }\nfunc (e *myEmitter) Emit(v T) { ... }\nfunc (e *myEmitter) AttachEstimator(est *WatermarkEstimator) { e.est = est }\nfunc (e *myEmitter) Value() typex.EventTime { return e.est.CurrentWatermark() }","handlingStrategy":"validation","validationCode":"for _, emit := range emitters {\n    if _, ok := emit.(exec.ReusableTimestampObservingWatermarkEmitter); !ok {\n        return fmt.Errorf(\"emitter %T does not implement ReusableTimestampObservingWatermarkEmitter; regenerate shims\", emit)\n    }\n}","typeGuard":"func isReusableWatermarkEmitter(e interface{}) bool {\n    _, ok := e.(exec.ReusableTimestampObservingWatermarkEmitter)\n    return ok\n}","tryCatchPattern":"if err := plan.Execute(ctx, id, mgr); err != nil {\n    if strings.Contains(err.Error(), \"ReusableTimestampObservingWatermarkEmitter\") {\n        return fmt.Errorf(\"regenerate shims with the beam code generator: %w\", err)\n    }\n}","preventionTips":["Regenerate code-shims whenever upgrading the Beam Go SDK version.","Run go generate ./... as part of the build after SDK upgrades.","If writing custom emitters, implement the full ReusableTimestampObservingWatermarkEmitter interface (AttachEstimator, Value)."],"tags":["go","apache-beam","codegen","interface","shims"],"backgroundTag":"deprecated-api-usage","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}