{"record":{"id":"7a7daa8d597eadd0","repo":"risingwavelabs/risingwave","slug":"either-topic-or-topic-field-must-be-set","errorCode":null,"errorMessage":"either topic or topic.field must be set","messagePattern":"either topic or topic\\.field must be set","errorType":"validation","errorClass":"SinkError::Config","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/mqtt.rs","lineNumber":202,"sourceCode":"\nimpl Sink for MqttSink {\n    type LogSinker = AsyncTruncateLogSinkerOf<MqttSinkWriter>;\n\n    const SINK_NAME: &'static str = MQTT_SINK;\n\n    crate::impl_validate_sink_unknown_fields!();\n\n    async fn validate(&self) -> Result<()> {\n        if !self.is_append_only {\n            return Err(SinkError::Mqtt(anyhow!(\n                \"MQTT sink only supports append-only mode\"\n            )));\n        }\n\n        if let Some(field) = &self.config.topic_field {\n            let _ = get_topic_field_index_path(&self.schema, field.as_str())?;\n        } else if self.config.topic.is_none() {\n            return Err(SinkError::Config(anyhow!(\n                \"either topic or topic.field must be set\"\n            )));\n        }\n\n        let _client = (self.config.common.build_client(0.into(), 0))\n            .context(\"validate mqtt sink error\")\n            .map_err(SinkError::Mqtt)?;\n\n        Ok(())\n    }\n\n    async fn new_log_sinker(&self, writer_param: SinkWriterParam) -> Result<Self::LogSinker> {\n        Ok(MqttSinkWriter::new(\n            self.config.clone(),\n            self.schema.clone(),\n            &self.format_desc,\n            &self.name,\n            writer_param.sink_id,","sourceCodeStart":184,"sourceCodeEnd":220,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/mqtt.rs#L184-L220","documentation":"Validation in MqttSink::validate: the MQTT topic target must be specified either as a static `topic` option or dynamically via `topic.field`; neither was set, so the writer would have nowhere to publish. Fires at sink creation/ALTER after the append-only and topic-field checks.","triggerScenarios":"Thrown at src/connector/src/sink/mqtt.rs:202 when the library encounters an invalid state.","commonSituations":"See trigger scenarios.","solutions":["Set a static 'topic' in the WITH options, or","Specify 'topic.field' referencing an existing column/field path in the sink schema.","Run the sink validation again after fixing the properties."],"exampleFix":null,"handlingStrategy":"validation","validationCode":null,"typeGuard":null,"tryCatchPattern":null,"preventionTips":[],"tags":[],"backgroundTag":null,"analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-23T08:17:48.524Z"}