{"record":{"id":"657cfa32ce35ffc8","repo":"risingwavelabs/risingwave","slug":"the-proto-payload-is-empty","errorCode":null,"errorMessage":"The proto payload is empty","messagePattern":"The proto payload is empty","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/connector/src/parser/protobuf/parser.rs","lineNumber":222,"sourceCode":"/// | 0          | 1-4        | 5-x             | x+1-end\n/// | magic-byte | schema-id  | message-indexes | protobuf-payload\npub(crate) fn resolve_pb_header(payload: &[u8]) -> ConnectorResult<&[u8]> {\n    // there's a message index array at the front of payload\n    // if it is the first message in proto def, the array is just and `0`\n    let (_, remained) = extract_schema_id(payload)?;\n    // The message indexes are encoded as int using variable-length zig-zag encoding,\n    // prefixed by the length of the array.\n    // Note that if the first byte is 0, it is equivalent to (1, 0) as an optimization.\n    match remained.first() {\n        Some(0) => Ok(&remained[1..]),\n        Some(_) => {\n            let (index_len, mut offset) = decode_varint_zigzag(remained)?;\n            for _ in 0..index_len {\n                offset += decode_varint_zigzag(&remained[offset..])?.1;\n            }\n            Ok(&remained[offset..])\n        }\n        None => bail!(\"The proto payload is empty\"),\n    }\n}\n\n#[cfg(test)]\nmod test {\n    use super::*;\n\n    #[test]\n    fn test_decode_varint_zigzag() {\n        // 1. Positive number\n        let buffer = vec![0x02];\n        let (value, len) = decode_varint_zigzag(&buffer).unwrap();\n        assert_eq!(value, 1);\n        assert_eq!(len, 1);\n\n        // 2. Negative number\n        let buffer = vec![0x01];\n        let (value, len) = decode_varint_zigzag(&buffer).unwrap();","sourceCodeStart":204,"sourceCodeEnd":240,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/parser/protobuf/parser.rs#L204-L240","documentation":"`resolve_pb_header` uses the encoded proto header to compute field-index offsets (Confluent wire format navigation). If header resolution yields no remaining payload bytes, there is nothing to decode, so the function bails. This indicates a malformed or empty protobuf message body rather than a schema problem.","triggerScenarios":"Called from `generate_accessor` while building the protobuf accessor; fires when the remaining payload slice after varint index navigation is `None` — i.e. the message bytes are empty (zero-length Kafka value) or nothing remains after the header.","commonSituations":"Kafka topic contains tombstone records (null/empty values) with protobuf encoding; a producer publishing empty payloads by misconfiguration; compacted-topic cleanup sending empty bodies; reading the wrong topic where records are not protobuf-encoded.","solutions":["Filter out empty/tombstone messages upstream or enable tombstone handling so empty payloads are skipped.","Verify the producer is publishing valid non-empty protobuf bytes.","Check you are reading the correct topic/partition with actual protobuf-encoded records.","Add a payload-size/type check in the producing application before send."],"exampleFix":"// before: producer sends empty payload\nproducer.send(record_with(b\"\"));\n// after: skip tombstones\nif payload.is_empty() { return Ok(()); }","handlingStrategy":"try-catch","validationCode":"// Producer-side guard before publishing:\nif payload.is_empty() { return Err(\"refusing to publish empty protobuf payload\"); }","typeGuard":null,"tryCatchPattern":"match decode_record(bytes) {\n    Err(e) if e.to_string().contains(\"The proto payload is empty\") => {\n        // treat as tombstone: skip or emit a delete marker\n        Ok(None)\n    }\n    other => other,\n}","preventionTips":["Skip or handle Kafka tombstone records (null values) in sources.","Validate payloads are non-empty in the producing application.","Verify the topic actually contains protobuf-encoded records."],"tags":["protobuf","kafka","empty-payload","cdc"],"backgroundTag":"empty-required-field","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"}