{"record":{"id":"117447bd302a6a92","repo":"risingwavelabs/risingwave","slug":"decode-pk-index-sink-position-delete-merger-metada","errorCode":null,"errorMessage":"decode pk-index sink position-delete merger metadata","messagePattern":"decode pk-index sink position-delete merger metadata","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/meta/src/manager/iceberg_pk_index_sink/coordinator.rs","lineNumber":596,"sourceCode":"            .ok()\n            .filter(|r| !matches!(r, PbIcebergPkIndexSinkRole::Unspecified))\n            .ok_or_else(|| anyhow!(\"iceberg pk-index sink report has invalid role: {}\", r.role))?;\n\n        match role {\n            PbIcebergPkIndexSinkRole::Writer => {\n                let commit_result = IcebergCommitResult::try_from(meta)?;\n                align_report_id(\n                    commit_result.schema_id,\n                    commit_result.partition_spec_id,\n                    &mut shared_schema_id,\n                    &mut shared_partition_spec_id,\n                )?;\n                data_files.extend(commit_result.data_files);\n            }\n            PbIcebergPkIndexSinkRole::PositionDeleteMerger => {\n                let commit_result =\n                    IcebergPositionDeleteCommitResult::try_from(meta).map_err(|e| {\n                        anyhow!(e).context(\"decode pk-index sink position-delete merger metadata\")\n                    })?;\n                align_report_id(\n                    commit_result.schema_id,\n                    commit_result.partition_spec_id,\n                    &mut shared_schema_id,\n                    &mut shared_partition_spec_id,\n                )?;\n                delete_files.extend(commit_result.delete_files);\n                overwrite_files.extend(commit_result.overwrite_files);\n            }\n            _ => unreachable!(),\n        }\n    }\n\n    Ok(IcebergPkIndexSinkAggResult {\n        schema_id: shared_schema_id.unwrap(),\n        partition_spec_id: shared_partition_spec_id.unwrap(),\n        data_files,","sourceCodeStart":578,"sourceCodeEnd":614,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/manager/iceberg_pk_index_sink/coordinator.rs#L578-L614","documentation":"A PositionDeleteMerger worker's commit report carries opaque metadata bytes that the coordinator decodes into IcebergPositionDeleteCommitResult via try_from(meta). When those bytes cannot be decoded (wrong message type, truncated, or an incompatible schema/version), aggregate_reports wraps the failure with this context and fails the pre-commit phase.","triggerScenarios":"pre_commit_epoch -> aggregate_reports handles a report with role PositionDeleteMerger and calls IcebergPositionDeleteCommitResult::try_from(meta), which errors because the report's meta bytes are not a valid position-delete commit-result payload.","commonSituations":"Version skew between meta and workers producing different protobuf layouts; a Writer-role payload delivered under the PositionDeleteMerger role; corrupted or replayed report metadata after a meta restart.","solutions":["Align meta and all worker binaries to the same version so the position-delete commit-result protobuf schema matches.","Inspect the raw report metadata for the failing sink to verify the payload type is IcebergPositionDeleteCommitResult.","Restart the sink/fragment so workers rebuild fresh reports instead of replaying stale metadata.","Check the producing worker to confirm it serializes the position-delete result, not the writer result."],"exampleFix":"// before\nlet commit_result = IcebergPositionDeleteCommitResult::try_from(meta)?;\n// after (fail with diagnosis)\nlet commit_result = IcebergPositionDeleteCommitResult::try_from(meta.clone())\n    .context(\"report meta is not IcebergPositionDeleteCommitResult; check worker/meta version alignment\")?;","handlingStrategy":"validation","validationCode":"// before trusting report meta\nif report.meta.is_none() || report.meta.as_ref().unwrap().is_empty() {\n    return Err(anyhow!(\"position-delete merger report meta is empty\"));\n}","typeGuard":"fn decodable_position_delete_meta(meta: &[u8]) -> bool {\n    IcebergPositionDeleteCommitResult::try_from(meta.to_vec()).is_ok()\n}","tryCatchPattern":"// caller\nmatch coordinator.pre_commit_epoch(epoch).await {\n    Err(e) if e.to_string().contains(\"position-delete merger metadata\") => {\n        // version skew / bad payload: log and re-register the sink\n    }\n    other => other?,\n}","preventionTips":["Keep meta and worker protobuf definitions from the same generated source.","Verify the report role matches the payload type before serializing.","Avoid replaying persisted report bytes across version upgrades."],"tags":["protobuf","iceberg","decode"],"backgroundTag":"protobuf-unmarshal-failed","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}