{"record":{"id":"a8ff7a824d25c7ea","repo":"risingwavelabs/risingwave","slug":"topic-field-not-found","errorCode":null,"errorMessage":"topic field `{}`  not found","messagePattern":"topic field `(.+?)`  not found","errorType":"validation","errorClass":"SinkError::Config","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/mqtt.rs","lineNumber":474,"sourceCode":"                        st.iter().enumerate().find(|(_, (s, _))| *s == field).map(\n                            |(pos, (_, dt))| {\n                                path.push(pos);\n                                dt\n                            },\n                        )\n                    }\n                    _ => None,\n                })\n            });\n\n    match dt {\n        Some(DataType::Varchar) => Ok(path),\n        Some(dt) => Err(SinkError::Config(anyhow!(\n            \"topic field `{}` must be of type string but got {:?}\",\n            topic_field,\n            dt\n        ))),\n        None => Err(SinkError::Config(anyhow!(\n            \"topic field `{}`  not found\",\n            topic_field\n        ))),\n    }\n}\n\n#[cfg(test)]\nmod test {\n    use risingwave_common::array::{DataChunk, DataChunkTestExt, RowRef};\n    use risingwave_common::catalog::{Field, Schema};\n    use risingwave_common::types::{DataType, StructType};\n\n    use super::{get_topic_field_index_path, get_topic_from_index_path};\n\n    #[test]\n    fn test_single_field_extraction() {\n        let schema = Schema::new(vec![Field::with_name(DataType::Varchar, \"topic\")]);\n        let path = get_topic_field_index_path(&schema, \"topic\").unwrap();","sourceCodeStart":456,"sourceCodeEnd":492,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/mqtt.rs#L456-L492","documentation":"`get_topic_field_index_path` cannot find the column named by the `topic_field` option in the sink's schema, so it cannot resolve an index path into the message. This error means the configured topic field does not exist on the data being sunk.","triggerScenarios":"Creating/validating an MQTT sink where the `topic_field` option names a column absent from the sink's schema (typo, wrong case, or column not included in the sink's SELECT).","commonSituations":"Typo in the `topic_field` value; user assumes a column exists in the sinked data but the SELECT/project excludes it; upstream rename of the column.","solutions":["Check the sink schema and correct the `topic_field` name to match an existing column exactly","If the column is not in the sinked stream, add it to the SELECT feeding the sink","Re-run `CREATE SINK` after fixing the option"],"exampleFix":"// before\nCREATE SINK s FROM t WITH (connector='mqtt', topic_field='devce_id');\n// after\nCREATE SINK s FROM t WITH (connector='mqtt', topic_field='device_id');","handlingStrategy":"validation","validationCode":"// ensure topic_field exists in the sink schema\nassert!(\n    schema.fields().iter().any(|f| f.name.as_str() == topic_field),\n    \"topic_field '{}' not found in schema {:?}\", topic_field, schema.fields()\n);","typeGuard":"fn topic_field_exists(schema: &Schema, name: &str) -> bool {\n    schema.fields().iter().any(|f| f.name.as_str() == name)\n}","tryCatchPattern":null,"preventionTips":["Match option values to column names exactly (case-sensitive)","Include the topic column in the sink's SELECT projection","Use DESC/SHOW to confirm the sinked schema first"],"tags":["mqtt","sink","config","missing-field"],"backgroundTag":"missing-required-config-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"}