apache/beam · critical

error decoding watermarks

Error message

error decoding watermarks

What it means

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.

Solutions

1. 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
2. Ensure the SDK harness container version matches the prism runner version (version skew causes coder mismatches)
3. Log and dump the failing bundle's raw bytes to identify which PCollection has the corrupt stream
4. Update to the latest Apache Beam release; several windowed-header decoding bugs have been fixed
5. File an issue with the pipeline and coder definitions if the bytes look correct

Example fix

// before: panic on any non-EOF decode error
if err != nil {
    if err == io.EOF { break }
    panic("error decoding watermarks")
}
// after: capture more diagnostics before failing
ws, et, pn, err := exec.DecodeWindowedValueHeader(info.WDec, tee)
if err != nil {
    if err == io.EOF { break }
    slog.Error("PersistBundle: error decoding watermarks", "error", err, "bundle", rb, "output", output)
    return fmt.Errorf("decoding windowed value header for bundle %v output %v: %w", rb, output, err)
}

// Before running: verify coder configuration
if !strings.Contains(coder.Ref, "windowed") || windowCoder == nil {
    return fmt.Errorf("PCollection %s missing window coder", pcolID)
}

When it happens

Trigger: 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.

Common situations: 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.

Understand the failure class

Background: "cannot parse invalid wire-format data", "cannot unmarshal", "failed unmarshalling": protobuf unmarshal errors explained — this error's family across 10 libraries.

Related errors

• couldn't decode characteristic for variant
• error decoding append bag user state window key
• error decoding residual header:
• error re-encoding characteristic for variant
• generating bundle for stage

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

Appendix: source

Thrown at sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go:899

	for output, data := range d.Raw {
		info := col2Coders[output]
		var newPending []element
		slog.Debug("PersistBundle: processing output", "bundle", rb, slog.String("output", output))
		for _, datum := range data {
			buf := bytes.NewBuffer(datum)
			if len(datum) == 0 {
				panic(fmt.Sprintf("zero length data for %v: ", output))
			}
			for {
				var rawBytes bytes.Buffer
				tee := io.TeeReader(buf, &rawBytes)
				ws, et, pn, err := exec.DecodeWindowedValueHeader(info.WDec, tee)
				if err != nil {
					if err == io.EOF {
						break
					}
					slog.Error("PersistBundle: error decoding watermarks", "error", err, "bundle", rb, slog.String("output", output))
					panic("error decoding watermarks")
				}
				if len(ws) == 0 {
					slog.Warn("PersistBundle: sdk provided a windowed value header 0 windows", "bundle", rb)
				}
				// TODO: Optimize unnecessary copies. This is doubleteeing.
				elmBytes := info.EDec(tee)
				var keyBytes []byte
				if info.KeyDec != nil {
					kbuf := bytes.NewBuffer(elmBytes)
					keyBytes = info.KeyDec(kbuf) // TODO: Optimize unnecessary copies. This is tripleteeing?
				}
				for _, w := range ws {
					newPending = append(newPending,
						element{
							window:    w,
							timestamp: et,
							pane:      stage.kind.getPaneOrDefault(stage, pn, w, keyBytes, rb.BundleID),
							elmBytes:  elmBytes,

View on GitHub (pinned to 12126d8942)