{"record":{"id":"801d7f1a1a1a53fb","repo":"risingwavelabs/risingwave","slug":"failed-to-parse-message-indexes","errorCode":null,"errorMessage":"failed to parse message indexes","messagePattern":"failed to parse message indexes","errorType":"exception","errorClass":"WireFormatError::ParseMessageIndexes","httpStatus":null,"severity":"error","filePath":"src/connector/src/schema/schema_registry/util.rs","lineNumber":52,"sourceCode":"    }\n    if urls.is_empty() {\n        bail_invalid_option_error!(\"no valid url provided, errs: {errs:?}\");\n    }\n    tracing::debug!(\n        \"schema registry client will use url {:?} to connect, the rest failed because: {:?}\",\n        urls,\n        errs\n    );\n    Ok(urls)\n}\n\n#[derive(Debug, thiserror::Error)]\npub enum WireFormatError {\n    #[error(\"failed to match the magic byte 0\")]\n    NoMagic,\n    #[error(\"failed to read the 4-byte schema ID\")]\n    NoSchemaId,\n    #[error(\"failed to parse message indexes\")]\n    ParseMessageIndexes,\n}\n\n/// Returns `(schema_id, payload)`\n///\n/// Refer to [Confluent schema registry wire format](https://docs.confluent.io/platform/7.6/schema-registry/fundamentals/serdes-develop/index.html#wire-format)\n///\n/// | Bytes | Area        | Description                                                                                        |\n/// |-------|-------------|----------------------------------------------------------------------------------------------------|\n/// | 0     | Magic Byte  | Confluent serialization format version number; currently always `0`.                               |\n/// | 1-4   | Schema ID   | 4-byte schema ID as returned by Schema Registry.                                                   |\n/// | 5-... | Data        | Serialized data for the specified schema format (for example, binary encoding for Avro or Protobuf.|\npub(crate) fn extract_schema_id(payload: &[u8]) -> Result<(i32, &[u8]), WireFormatError> {\n    use byteorder::{BigEndian, ReadBytesExt as _};\n\n    let mut cursor = payload;\n    if !cursor.read_u8().is_ok_and(|magic| magic == 0) {\n        return Err(WireFormatError::NoMagic);","sourceCodeStart":34,"sourceCodeEnd":70,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/schema/schema_registry/util.rs#L34-L70","documentation":"WireFormatError::ParseMessageIndexes is thrown when decoding a Confluent wire format v1 (multi-message/array) payload where the trailing array of message indexes cannot be parsed. Indexes are used to point at a nested position in the schema; failure means the variable-length index encoding is malformed.","triggerScenarios":"Parsing a wire-format payload whose trailing bytes (after magic + schema ID) should be a zigzag-varint-encoded index array but are missing, empty, or corrupt — e.g. truncated messages or non-conformant producers.","commonSituations":"Custom producers writing v1 wire format incorrectly, truncated Kafka records, or payloads hand-assembled in tests without proper index encoding.","solutions":["Verify the producer's Confluent serializer version and that it writes indexes per the wire format spec.","Dump raw payload bytes and validate the trailing zigzag varint index array manually.","Fall back to producing wire-format v0 messages (single top-level record) if nested indexing is not needed.","Re-serialize affected records with the official Confluent serializer."],"exampleFix":"// before: assuming v0 payload\nlet (id, payload) = parse(bytes)?; // indexes never parsed\n// after: handle v1 with valid index array\nif bytes.first() == Some(&1) {\n    let indexes = parse_message_indexes(&mut cursor)?; // zigzag varints\n}","handlingStrategy":"try-catch","validationCode":null,"typeGuard":"fn is_wire_format_v1(b: &[u8]) -> bool { b.first() == Some(&1) }","tryCatchPattern":"match parse_wire_format(bytes) {\n    Ok((id, payload)) => decode(id, payload),\n    Err(WireFormatError::ParseMessageIndexes) => {\n        tracing::warn!(\"bad v1 message indexes; falling back to raw decode\");\n        fallback_decode(bytes);\n    }\n    Err(e) => return Err(e.into()),\n}","preventionTips":["Use official Confluent serializers that emit spec-compliant v1 indexes.","Prefer v0 wire format when nested schemas are not needed.","Validate index encoding with unit tests on sample payloads.","Guard against truncated records before index parsing."],"tags":["kafka","wire-format","protobuf"],"backgroundTag":"unexpected-response-shape","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"}