{"record":{"id":"f96191e96233ae36","repo":"risingwavelabs/risingwave","slug":"invalid-commit-metadata-missing-snapshot-id","errorCode":null,"errorMessage":"Invalid commit metadata: missing snapshot_id","messagePattern":"Invalid commit metadata: missing snapshot_id","errorType":"exception","errorClass":"SinkError::Iceberg","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/iceberg/commit.rs","lineNumber":498,"sourceCode":"                sink_id = %self.sink_id,\n                table = %self.table.identifier(),\n                epoch,\n                \"iceberg_sink_commit_skipped_empty_metadata\",\n            );\n            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\",","sourceCodeStart":480,"sourceCodeEnd":516,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/iceberg/commit.rs#L480-L516","documentation":"After popping the last payload element, commit_data treats it as the little-endian snapshot_id bytes. If the payload list was empty at that point (only reachable when the earlier is_empty guard passed but pop still fails — effectively an invariant race), it returns this error. The snapshot id is mandatory to create the new Iceberg snapshot.","triggerScenarios":"commit_data with a payload that has no trailing snapshot_id element — a metadata blob assembled without appending the i64 LE snapshot id.","commonSituations":"Custom or outdated code paths assembling commit metadata without the snapshot id suffix; corrupted state where the last element was dropped.","solutions":["Ensure the metadata producer always appends snapshot_id.to_le_bytes() as the final payload element","Align the writer's serialization format with commit_data's expected layout","Regenerate the commit metadata by restarting the sink epoch"],"exampleFix":"// before (payload without snapshot id)\npayload = [write_result_1_bytes]\n// after\npayload = [write_result_1_bytes, snapshot_id.to_le_bytes()]","handlingStrategy":"validation","validationCode":"fn payload_has_snapshot_id(payload: &[Vec<u8>]) -> bool {\n    payload.len() >= 1 && payload.last().map_or(false, |b| b.len() == 8)\n}","typeGuard":"fn as_le_i64(bytes: &[u8]) -> Option<i64> {\n    i64::from_le_bytes(bytes.try_into().ok()?).into()\n}","tryCatchPattern":"let Some(snapshot_id_bytes) = payload.pop() else {\n    return Err(SinkError::Iceberg(anyhow!(\"missing snapshot_id\")));\n};","preventionTips":["Always append snapshot_id.to_le_bytes() as the final payload element","Document the metadata layout next to the serializer","Round-trip test the payload assembly/parsing pair"],"tags":["rust","iceberg","commit","missing-field"],"backgroundTag":"missing-required-argument","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"}