{"record":{"id":"07fce8613a16f8b5","repo":"risingwavelabs/risingwave","slug":"mqtt-error-0","errorCode":null,"errorMessage":"Mqtt error: {0}","messagePattern":"Mqtt error: (.+?)","errorType":"exception","errorClass":"SinkError","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/mod.rs","lineNumber":1136,"sourceCode":"        anyhow::Error,\n    ),\n    #[error(\"coordinator error: {0}\")]\n    Coordinator(\n        #[source]\n        #[backtrace]\n        anyhow::Error,\n    ),\n    #[error(\"ClickHouse error: {0}\")]\n    ClickHouse(String),\n    #[error(\"Redis error: {0}\")]\n    Redis(String),\n    #[error(\"Http error: {0}\")]\n    Http(\n        #[source]\n        #[backtrace]\n        anyhow::Error,\n    ),\n    #[error(\"Mqtt error: {0}\")]\n    Mqtt(\n        #[source]\n        #[backtrace]\n        anyhow::Error,\n    ),\n    #[error(\"Nats error: {0}\")]\n    Nats(\n        #[source]\n        #[backtrace]\n        anyhow::Error,\n    ),\n    #[error(\"Google Pub/Sub error: {0}\")]\n    GooglePubSub(\n        #[source]\n        #[backtrace]\n        anyhow::Error,\n    ),\n    #[error(\"Doris/Starrocks connect error: {0}\")]","sourceCodeStart":1118,"sourceCodeEnd":1154,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/mod.rs#L1118-L1154","documentation":"SinkError::Mqtt(anyhow::Error) in src/connector/src/sink/mod.rs:1137 wraps failures from the MQTT sink connector. It preserves the source anyhow::Error and backtrace from the underlying MQTT client (rumqttc). Raised when connecting to the broker, authenticating, or publishing messages fails.","triggerScenarios":"CREATE SINK ... WITH (connector='mqtt') when: the broker address is unreachable, the TCP/TLS connection fails, CONNECT is rejected (bad client id, credentials, or not authorized), the publish packet fails (QoS 1/2 no PUBACK, queue full, or client disconnected).","commonSituations":"Wrong broker host/port or broker down; MQTT credentials rotated and now rejected; TLS to a broker with a self-signed cert without CA configuration; broker session/queue limits exceeded under high throughput causing EventLoop disconnections.","solutions":["Test broker reachability (mosquitto_pub -h <host> -p <port> -u <user> -P <pass> -t test -m hi) and fix the broker URL/credentials in the sink options.","Check TLS/CA settings in the sink definition against the broker's certificate setup.","Verify the client id is unique and the account is authorized for the target topic (broker ACLs).","Lower QoS or increase broker queue/session limits if publishes fail under load; check broker logs for disconnect reasons.","Read the wrapped rumqttc error in RisingWave logs to distinguish connection vs publish-stage failures."],"exampleFix":"// before: credentials rejected at CONNECT\nCREATE SINK mqtt_sink FROM mv WITH (\n  connector = 'mqtt',\n  url = 'mqtt://broker:1883',\n  username = 'rw', password = 'old-pass'\n);\n// after: rotated credentials supplied\nCREATE SINK mqtt_sink FROM mv WITH (\n  connector = 'mqtt',\n  url = 'mqtt://broker:1883',\n  username = 'rw', password = 'new-pass'\n);","handlingStrategy":"retry","validationCode":"# Verify broker connectivity and auth before creating the sink\nmosquitto_pub -h \"$MQTT_HOST\" -p \"$MQTT_PORT\" -u \"$MQTT_USER\" -P \"$MQTT_PASS\" -t 'rw/health' -m 'check' && echo OK\n# or in Rust\nlet opts = rumqttc::MqttOptions::new(\"rw-check\", host, port);\n// attempt connect with a short timeout before sinking","typeGuard":"fn is_valid_mqtt_url(url: &str) -> bool {\n    url.starts_with(\"mqtt://\") || url.starts_with(\"mqtts://\") || url.starts_with(\"ssl://\")\n}","tryCatchPattern":"// Distinguish connect-stage from publish-stage failures\nmatch sink_result {\n    Err(SinkError::Mqtt(e)) if is_connect_error(&e) => retry_connect_with_backoff(),\n    Err(SinkError::Mqtt(e)) => { log(&e); /* check broker logs, QoS, queue limits */ }\n    Ok(v) => process(v),\n}","preventionTips":["Validate broker host/port/credentials with mosquitto_pub before creating the sink","Use a unique client id per sink instance to avoid session kicks","Configure TLS CA material when the broker requires it (mqtts://)","Match QoS to broker capabilities and size the broker session/queue for peak throughput","Watch broker disconnect logs and RisingWave sink error metrics"],"tags":["mqtt","sink","network","risingwave-connector"],"backgroundTag":"network-request-failed","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"}