{"record":{"id":"9792dc7b7c24c1ec","repo":"risingwavelabs/risingwave","slug":"failed-to-connect-to-mqtt-broker","errorCode":null,"errorMessage":"Failed to connect to mqtt broker","messagePattern":"Failed to connect to mqtt broker","errorType":"exception","errorClass":"ConnectorError","httpStatus":null,"severity":"error","filePath":"src/connector/src/source/mqtt/enumerator/mod.rs","lineNumber":152,"sourceCode":"        // connection_check always gets created if it doesn't exist\n        let connection_check = connection_check.unwrap();\n\n        Ok(Self {\n            topic: properties.topic,\n            broker: broker_url,\n            connection_check,\n        })\n    }\n\n    async fn list_splits(&mut self) -> ConnectorResult<Vec<MqttSplit>> {\n        if !self.connection_check.is_connected() {\n            let start = std::time::Instant::now();\n            loop {\n                if self.connection_check.is_connected() {\n                    break;\n                };\n                if start.elapsed().as_secs() > 10 {\n                    bail!(\"Failed to connect to mqtt broker\");\n                }\n\n                tokio::time::sleep(std::time::Duration::from_millis(500)).await;\n            }\n        }\n        tracing::debug!(\"found new splits {} for broker {}\", self.topic, self.broker);\n        Ok(vec![MqttSplit::new(self.topic.clone())])\n    }\n}\n","sourceCodeStart":134,"sourceCodeEnd":162,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/source/mqtt/enumerator/mod.rs#L134-L162","documentation":"The MQTT source enumerator's `list_splits` waits (polling every 500ms) up to 10 seconds for the background MQTT connection to become established, checking `self.connection_check.is_connected()`. If the broker is still not connected after 10 seconds, `bail!` aborts with \"Failed to connect to mqtt broker\".","triggerScenarios":"Starting an MQTT source where the broker address is wrong/unreachable, TLS handshake fails, credentials are rejected, or the broker is simply slower than 10 seconds to accept the connection — any case where `connection_check` stays disconnected past the 10s deadline.","commonSituations":"Misconfigured broker URL or port in WITH options; firewall/security group blocking the MQTT port; broker down or restarting; auth (username/password/TLS cert) mismatch; slow networks or heavily loaded brokers exceeding the fixed 10s timeout.","solutions":["Verify the broker host/port in the source WITH options (e.g. `broker = 'mqtt://host:1883'`) and test reachability from the RisingWave host (`nc -vz host 1883`).","Check MQTT credentials and TLS settings (ca.pem/cert/key options) are correct.","Confirm the broker is running and accepting connections (mosquitto status / broker logs).","Restart the source after the broker is reachable; the check runs again on each `list_splits`.","If the broker is merely slow (>10s to accept), reduce load or increase availability of the broker — the timeout is hardcoded."],"exampleFix":"// before\nWITH (\n  connector = 'mqtt',\n  broker = 'mqtt://broker.internal:8883'\n)  -- TLS configured nowhere; handshake hangs >10s\n// after\nWITH (\n  connector = 'mqtt',\n  broker = 'mqtt://broker.internal:8883',\n  username = 'rw_user',\n  password = '***',\n  ca_certificate = '...'\n)","handlingStrategy":"retry","validationCode":"// Probe broker reachability and TLS before creating the source\nuse tokio::net::TcpStream;\nlet addr = \"broker.internal:1883\";\nTcpStream::connect(addr).await\n    .context(format!(\"MQTT broker unreachable at {addr}; fix broker option or network before creating the source\"))?;","typeGuard":null,"tryCatchPattern":"// Retry source creation with backoff while the broker is down\nfor attempt in 0..5 {\n    match create_mqtt_source(opts.clone()).await {\n        Ok(_) => break,\n        Err(e) if e.to_string().contains(\"Failed to connect to mqtt broker\") && attempt < 4 => {\n            tokio::time::sleep(Duration::from_secs(15 * (attempt + 1))).await;\n        }\n        Err(e) => return Err(e),\n    }\n}","preventionTips":["Validate broker host/port/network reachability (firewall, security groups) before creating the source.","Double-check username/password and TLS certificate options against broker config.","Ensure the broker is running and sized to accept connections within ~10 seconds.","Monitor broker health and restarts; treat this error as a broker availability signal."],"tags":["mqtt","connection-timeout","broker","network"],"backgroundTag":"connection-refused","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"}