risingwavelabs/risingwave · error

invalid message type

Error message

invalid message type

What it means

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.

Source

Thrown at src/connector/src/source/cdc/source/message.rs:181

        msg_type: cdc_message::CdcMessageType,
        source_type: SourceType,
    ) -> Self {
        let (db_name_end, table_name_start) =
            Self::derive_name_indices_from_full_table_name(&full_table_name, source_type);
        Self {
            source_type,
            db_name_end,
            table_name_start,
            full_table_name,
            source_ts_ms,
            msg_type: msg_type.into(),
        }
    }
}

impl From<CdcMessage> for SourceMessage {
    fn from(message: CdcMessage) -> Self {
        let msg_type = message.get_msg_type().expect("invalid message type");
        let source_type = message.get_source_type().unwrap_or(SourceType::Unspecified);
        SourceMessage {
            key: if message.key.is_empty() {
                None // only data message has key
            } else {
                Some(message.key.as_bytes().to_vec())
            },
            payload: if message.payload.is_empty() {
                None // heartbeat message
            } else {
                Some(message.payload.as_bytes().to_vec())
            },
            offset: message.offset,
            split_id: message.partition.into(),
            meta: SourceMeta::DebeziumCdc(DebeziumCdcMeta::new(
                message.full_table_name,
                message.source_ts_ms,
                msg_type,

View on GitHub (pinned to 6469eb736d)

Solutions

1. Identify the upstream connector/version mismatch and align the message format (recreate the source after upgrade).
2. Restart/recreate the CDC source to rebuild state and re-snapshot if messages are corrupt.
3. 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.

Example fix

// before
let msg_type = message.get_msg_type().expect("invalid message type");
// after
let msg_type = message.get_msg_type().unwrap_or_default(); // or map_err into ConnectorResult and skip

fn has_valid_msg_type(m: &CdcMessage) -> bool {
    matches!(m.get_msg_type(), Ok(_))
}

let msg_type = message.get_msg_type().unwrap_or_else(|e| {
    tracing::warn!("skipping CdcMessage with invalid type: {e}");
    return; // skip message instead of panicking
});

When it happens

Trigger: 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.

Common situations: Upgrading RisingWave across versions where stored/sent message formats changed; Debezium emitting unexpected envelope message kinds; corrupted state after a crash.

Understand the failure class

Background: Invalid enum value errors: "Unknown type", "Invalid scope", "must be one of" — when a string is not on the library's allowed list — this error's family across 23 libraries.

Related errors

• implement MySQL CDC parallelized backfill
• Can't convert int {} to ScalarImpl::{}
• No result returned by `SELECT sys.fn_cdc_get_max_lsn()`
• All valid CDC connectors should have returned by now
• failed to convert JSON schema to Avro schema: {}

AI-assisted analysis of risingwavelabs/risingwave@6469eb736d (2026-09-11). Data as JSON: /api/errors/ca93ff54a6939fb3. Report an issue: GitHub.