{"record":{"id":"ca93ff54a6939fb3","repo":"risingwavelabs/risingwave","slug":"invalid-message-type","errorCode":null,"errorMessage":"invalid message type","messagePattern":"invalid message type","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/connector/src/source/cdc/source/message.rs","lineNumber":181,"sourceCode":"        msg_type: cdc_message::CdcMessageType,\n        source_type: SourceType,\n    ) -> Self {\n        let (db_name_end, table_name_start) =\n            Self::derive_name_indices_from_full_table_name(&full_table_name, source_type);\n        Self {\n            source_type,\n            db_name_end,\n            table_name_start,\n            full_table_name,\n            source_ts_ms,\n            msg_type: msg_type.into(),\n        }\n    }\n}\n\nimpl From<CdcMessage> for SourceMessage {\n    fn from(message: CdcMessage) -> Self {\n        let msg_type = message.get_msg_type().expect(\"invalid message type\");\n        let source_type = message.get_source_type().unwrap_or(SourceType::Unspecified);\n        SourceMessage {\n            key: if message.key.is_empty() {\n                None // only data message has key\n            } else {\n                Some(message.key.as_bytes().to_vec())\n            },\n            payload: if message.payload.is_empty() {\n                None // heartbeat message\n            } else {\n                Some(message.payload.as_bytes().to_vec())\n            },\n            offset: message.offset,\n            split_id: message.partition.into(),\n            meta: SourceMeta::DebeziumCdc(DebeziumCdcMeta::new(\n                message.full_table_name,\n                message.source_ts_ms,\n                msg_type,","sourceCodeStart":163,"sourceCodeEnd":199,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/source/cdc/source/message.rs#L163-L199","documentation":"Converting a `CdcMessage` into a `SourceMessage` calls `message.get_msg_type()` and panics with `.expect(\"invalid message type\")` if the message's type field cannot be decoded (e.g. protobuf enum value unrecognized). It indicates a malformed or incompatible CDC message entering the pipeline.","triggerScenarios":"A CdcMessage whose msg_type enum fails to map (unknown/unset protobuf enum value), typically from corrupt or truncated connector state, a version mismatch between message producers/consumers, or an upstream connector bug emitting messages without a valid type.","commonSituations":"Upgrading RisingWave across versions where stored/sent message formats changed; Debezium emitting unexpected envelope message kinds; corrupted state after a crash.","solutions":["Identify the upstream connector/version mismatch and align the message format (recreate the source after upgrade).","Restart/recreate the CDC source to rebuild state and re-snapshot if messages are corrupt.","Patch `From<CdcMessage> for SourceMessage` to return a typed error / skip-and-log instead of `expect` so a single bad message does not panic the source."],"exampleFix":"// before\nlet msg_type = message.get_msg_type().expect(\"invalid message type\");\n// after\nlet msg_type = message.get_msg_type().unwrap_or_default(); // or map_err into ConnectorResult and skip","handlingStrategy":"try-catch","validationCode":null,"typeGuard":"fn has_valid_msg_type(m: &CdcMessage) -> bool {\n    matches!(m.get_msg_type(), Ok(_))\n}","tryCatchPattern":"let msg_type = message.get_msg_type().unwrap_or_else(|e| {\n    tracing::warn!(\"skipping CdcMessage with invalid type: {e}\");\n    return; // skip message instead of panicking\n});","preventionTips":["Recreate sources after major version upgrades to avoid stale message formats.","Watch for connector crash loops and inspect upstream Debezium output on anomalies.","Patch the conversion to skip-and-log malformed messages."],"tags":["cdc","message-decoding","panic","enum"],"backgroundTag":"invalid-enum-value","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"}