{"record":{"id":"5a833949c3989d0f","repo":"risingwavelabs/risingwave","slug":"invalid-snapshot-id-bytes","errorCode":null,"errorMessage":"Invalid snapshot id bytes","messagePattern":"Invalid snapshot id bytes","errorType":"exception","errorClass":"SinkError::Iceberg","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/iceberg/commit.rs","lineNumber":503,"sourceCode":"            return Ok(());\n        }\n\n        // Deserialize commit metadata\n        let mut payload = deserialize_metadata(commit_metadata);\n        if payload.is_empty() {\n            return Err(SinkError::Iceberg(anyhow!(\n                \"Invalid commit metadata: empty payload\"\n            )));\n        }\n\n        // Last element is snapshot_id\n        let snapshot_id_bytes = payload.pop().ok_or_else(|| {\n            SinkError::Iceberg(anyhow!(\"Invalid commit metadata: missing snapshot_id\"))\n        })?;\n        let snapshot_id = i64::from_le_bytes(\n            snapshot_id_bytes\n                .try_into()\n                .map_err(|_| SinkError::Iceberg(anyhow!(\"Invalid snapshot id bytes\")))?,\n        );\n\n        // Remaining elements are write_results\n        let write_results = payload\n            .into_iter()\n            .map(|p| IcebergCommitResult::try_from_serialized_bytes(&p))\n            .collect::<Result<Vec<_>>>()?;\n\n        let snapshot_committed = self.is_snapshot_id_in_iceberg(snapshot_id).await?;\n\n        if snapshot_committed {\n            tracing::info!(\n                iceberg_component = \"sink_committer\",\n                iceberg_operation = \"commit\",\n                sink_id = %self.sink_id,\n                sink_name = %self.param.sink_name,\n                table = %self.table.identifier(),\n                epoch,","sourceCodeStart":485,"sourceCodeEnd":521,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/iceberg/commit.rs#L485-L521","documentation":"The popped snapshot_id bytes are converted to an 8-byte array with try_into() before i64::from_le_bytes; if the byte slice length differs from 8, this error fires. It protects against truncated or malformed snapshot id payloads.","triggerScenarios":"The final payload element is not exactly 8 bytes — e.g. snapshot id stored as big-endian with different handling, serialized as a decimal string, or the payload slice was truncated by storage.","commonSituations":"Corruption or truncation in the state store; mixed writer versions using a different snapshot id encoding; hand-assembled test metadata.","solutions":["Ensure snapshot_id is written exactly as i64::to_le_bytes() (8 bytes, little-endian)","Check writer/reader version compatibility for the metadata encoding","Inspect the offending payload length and regenerate the commit metadata"],"exampleFix":"// before\npayload.push(snapshot_id.to_be_bytes().to_vec());\n// after\npayload.push(snapshot_id.to_le_bytes().to_vec());","handlingStrategy":"validation","validationCode":"fn is_valid_snapshot_id_bytes(b: &[u8]) -> bool {\n    b.len() == 8\n}","typeGuard":"fn le_i64(b: &[u8]) -> Option<i64> {\n    b.try_into().ok().map(i64::from_le_bytes)\n}","tryCatchPattern":"let snapshot_id = i64::from_le_bytes(\n    snapshot_id_bytes.try_into()\n        .map_err(|_| SinkError::Iceberg(anyhow!(\"Invalid snapshot id bytes\")))?,\n);","preventionTips":["Encode snapshot ids strictly as i64::to_le_bytes","Verify byte lengths after reading from the state store","Keep encoding identical across writer versions"],"tags":["rust","iceberg","byte-format","truncated-data"],"backgroundTag":"invalid-argument-format","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-23T08:17:48.524Z"}