{"record":{"id":"7e40366f221888aa","repo":"risingwavelabs/risingwave","slug":"sinkerror-mqtt-anyhow-e","errorCode":null,"errorMessage":"SinkError::Mqtt(anyhow!(e))","messagePattern":"SinkError::Mqtt\\(anyhow!\\(e\\)\\)","errorType":"exception","errorClass":"SinkError::Mqtt","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/mqtt.rs","lineNumber":292,"sourceCode":"                _ => {\n                    return Err(SinkError::Config(anyhow!(\n                        \"mqtt sink encode unsupported: {:?}\",\n                        format_desc.encode,\n                    )));\n                }\n            },\n            _ => {\n                return Err(SinkError::Config(anyhow!(\n                    \"MQTT sink only supports append-only mode\"\n                )));\n            }\n        };\n        let qos = config.common.qos();\n\n        let (client, mut eventloop) = config\n            .common\n            .build_client(actor_id, sink_id.as_raw_id())\n            .map_err(|e| SinkError::Mqtt(anyhow!(e)))?;\n\n        let stopped = Arc::new(AtomicBool::new(false));\n        let stopped_clone = stopped.clone();\n        tokio::spawn(async move {\n            while !stopped_clone.load(std::sync::atomic::Ordering::Relaxed) {\n                match eventloop.poll().await {\n                    Ok(_) => (),\n                    Err(err) => match err {\n                        ConnectionError::Timeout(_) => (),\n                        ConnectionError::MqttState(rumqttc::v5::StateError::Io(err))\n                        | ConnectionError::Io(err)\n                            if err.kind() == std::io::ErrorKind::ConnectionAborted\n                                || err.kind() == std::io::ErrorKind::ConnectionReset =>\n                        {\n                            continue;\n                        }\n                        err => {\n                            tracing::error!(","sourceCodeStart":274,"sourceCodeEnd":310,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/mqtt.rs#L274-L310","documentation":"This error wraps any failure from the MQTT client builder (`build_client`) into a `SinkError::Mqtt` while constructing an MQTT sink writer in `MqttSinkWriter::new`. It means the MQTT sink could not create its client/eventloop (typically a connection or broker URL problem), so the sink cannot be created.","triggerScenarios":"Calling `MqttSinkWriter::new(actor_id, sink_id, config, ...)` when `MqttConfig::build_client` fails: malformed broker URL, unresolvable/invalid host, bad connection options, or missing required MQTT fields in the WITH options.","commonSituations":"Typo in the broker hostname or port in the sink WITH options; broker unreachable from the compute node; invalid URL scheme (e.g. missing tcp:// or ssl://); mqtts configured without proper TLS fields.","solutions":["Check the sink WITH options for a correct broker URL (scheme, host, port), e.g. `url='mqtt://broker:1883'`","Verify the broker is reachable from the RisingWave compute node (ping / telnet to host:port)","Re-run `CREATE SINK` with corrected options; the error message from the underlying client usually names the exact bad field","If using TLS/mqtts, ensure required TLS options (e.g. `tls_mode`, certs) are provided"],"exampleFix":"// before\nCREATE SINK s FROM t WITH (connector='mqtt', url='broker:1883');\n// after\nCREATE SINK s FROM t WITH (connector='mqtt', url='mqtt://broker:1883', qos='at_least_once');","handlingStrategy":"validation","validationCode":"// before CREATE SINK\nlet url = options.get(\"url\").expect(\"mqtt url required\");\nassert!(url.starts_with(\"mqtt://\") || url.starts_with(\"mqtts://\"), \"url must include scheme\");\nlet (host, port) = parse_host_port(&url).expect(\"valid host:port\");\ntokio::net::TcpStream::connect((host.as_str(), port)).await.expect(\"broker reachable\");","typeGuard":"fn is_valid_mqtt_url(u: &str) -> bool {\n    matches!(u.split_once(\"://\"), Some((\"mqtt\" | \"mqtts\", rest))) && !rest.is_empty()\n}","tryCatchPattern":null,"preventionTips":["Always include the scheme in the broker URL (mqtt:// or mqtts://)","Test broker connectivity from the compute node before creating the sink","Keep MQTT client option names consistent with MqttConfig fields"],"tags":["mqtt","sink","network","connection"],"backgroundTag":"connection-refused","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"}