{"record":{"id":"167b1ceee795c5fe","repo":"apache/beam","slug":"error-decoding-watermarks","errorCode":null,"errorMessage":"error decoding watermarks","messagePattern":"error decoding watermarks","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"critical","filePath":"sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go","lineNumber":899,"sourceCode":"\tfor output, data := range d.Raw {\n\t\tinfo := col2Coders[output]\n\t\tvar newPending []element\n\t\tslog.Debug(\"PersistBundle: processing output\", \"bundle\", rb, slog.String(\"output\", output))\n\t\tfor _, datum := range data {\n\t\t\tbuf := bytes.NewBuffer(datum)\n\t\t\tif len(datum) == 0 {\n\t\t\t\tpanic(fmt.Sprintf(\"zero length data for %v: \", output))\n\t\t\t}\n\t\t\tfor {\n\t\t\t\tvar rawBytes bytes.Buffer\n\t\t\t\ttee := io.TeeReader(buf, &rawBytes)\n\t\t\t\tws, et, pn, err := exec.DecodeWindowedValueHeader(info.WDec, tee)\n\t\t\t\tif err != nil {\n\t\t\t\t\tif err == io.EOF {\n\t\t\t\t\t\tbreak\n\t\t\t\t\t}\n\t\t\t\t\tslog.Error(\"PersistBundle: error decoding watermarks\", \"error\", err, \"bundle\", rb, slog.String(\"output\", output))\n\t\t\t\t\tpanic(\"error decoding watermarks\")\n\t\t\t\t}\n\t\t\t\tif len(ws) == 0 {\n\t\t\t\t\tslog.Warn(\"PersistBundle: sdk provided a windowed value header 0 windows\", \"bundle\", rb)\n\t\t\t\t}\n\t\t\t\t// TODO: Optimize unnecessary copies. This is doubleteeing.\n\t\t\t\telmBytes := info.EDec(tee)\n\t\t\t\tvar keyBytes []byte\n\t\t\t\tif info.KeyDec != nil {\n\t\t\t\t\tkbuf := bytes.NewBuffer(elmBytes)\n\t\t\t\t\tkeyBytes = info.KeyDec(kbuf) // TODO: Optimize unnecessary copies. This is tripleteeing?\n\t\t\t\t}\n\t\t\t\tfor _, w := range ws {\n\t\t\t\t\tnewPending = append(newPending,\n\t\t\t\t\t\telement{\n\t\t\t\t\t\t\twindow:    w,\n\t\t\t\t\t\t\ttimestamp: et,\n\t\t\t\t\t\t\tpane:      stage.kind.getPaneOrDefault(stage, pn, w, keyBytes, rb.BundleID),\n\t\t\t\t\t\t\telmBytes:  elmBytes,","sourceCodeStart":881,"sourceCodeEnd":917,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go#L881-L917","documentation":"Prism's element manager decodes a Beam windowed-value header from a bundle's teed data stream via DecodeWindowedValueHeader. This panic fires when the header bytes cannot be decoded into windows/key/partition and the error is not a clean io.EOF. It indicates corrupt, truncated, or mis-coded data between the SDK harness and the runner.","triggerScenarios":"DecodeWindowedValueHeader returns a non-EOF error while persisting a bundle — typically a coder mismatch between what the SDK declared and the bytes it sent, or a truncated stream. The code breaks cleanly on io.EOF but panics on any other decode error.","commonSituations":"Custom or misconfigured coders (e.g. a window coder not length-prefixed), Beam SDK/runner version skew between harness container and prism, corrupted data channels during harness instability.","solutions":["Verify the window and value coders declared by the pipeline match the SDK harness's actual encoding; force length-prefixed coders if using custom window coders","Ensure the SDK harness container version matches the prism runner version (version skew causes coder mismatches)","Log and dump the failing bundle's raw bytes to identify which PCollection has the corrupt stream","Update to the latest Apache Beam release; several windowed-header decoding bugs have been fixed","File an issue with the pipeline and coder definitions if the bytes look correct"],"exampleFix":"// before: panic on any non-EOF decode error\nif err != nil {\n    if err == io.EOF { break }\n    panic(\"error decoding watermarks\")\n}\n// after: capture more diagnostics before failing\nws, et, pn, err := exec.DecodeWindowedValueHeader(info.WDec, tee)\nif err != nil {\n    if err == io.EOF { break }\n    slog.Error(\"PersistBundle: error decoding watermarks\", \"error\", err, \"bundle\", rb, \"output\", output)\n    return fmt.Errorf(\"decoding windowed value header for bundle %v output %v: %w\", rb, output, err)\n}","handlingStrategy":"validation","validationCode":"// Before running: verify coder configuration\nif !strings.Contains(coder.Ref, \"windowed\") || windowCoder == nil {\n    return fmt.Errorf(\"PCollection %s missing window coder\", pcolID)\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Pin identical Beam versions for SDK harness images and the prism runner","Prefer standard windowing strategies and coders over custom ones","Add integration smoke tests exercising windowed pipelines before upgrades"],"tags":["go","beam","prism","decoding","panic"],"backgroundTag":"protobuf-unmarshal-failed","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"}