risingwavelabs/risingwave · error · SinkError::Mqtt

SinkError::Mqtt(anyhow!(e))

Error message

SinkError::Mqtt(anyhow!(e))

What it means

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.

Source

Thrown at src/connector/src/sink/mqtt.rs:292

                _ => {
                    return Err(SinkError::Config(anyhow!(
                        "mqtt sink encode unsupported: {:?}",
                        format_desc.encode,
                    )));
                }
            },
            _ => {
                return Err(SinkError::Config(anyhow!(
                    "MQTT sink only supports append-only mode"
                )));
            }
        };
        let qos = config.common.qos();

        let (client, mut eventloop) = config
            .common
            .build_client(actor_id, sink_id.as_raw_id())
            .map_err(|e| SinkError::Mqtt(anyhow!(e)))?;

        let stopped = Arc::new(AtomicBool::new(false));
        let stopped_clone = stopped.clone();
        tokio::spawn(async move {
            while !stopped_clone.load(std::sync::atomic::Ordering::Relaxed) {
                match eventloop.poll().await {
                    Ok(_) => (),
                    Err(err) => match err {
                        ConnectionError::Timeout(_) => (),
                        ConnectionError::MqttState(rumqttc::v5::StateError::Io(err))
                        | ConnectionError::Io(err)
                            if err.kind() == std::io::ErrorKind::ConnectionAborted
                                || err.kind() == std::io::ErrorKind::ConnectionReset =>
                        {
                            continue;
                        }
                        err => {
                            tracing::error!(

View on GitHub (pinned to 6469eb736d)

Solutions

1. Check the sink WITH options for a correct broker URL (scheme, host, port), e.g. `url='mqtt://broker:1883'`
2. Verify the broker is reachable from the RisingWave compute node (ping / telnet to host:port)
3. Re-run `CREATE SINK` with corrected options; the error message from the underlying client usually names the exact bad field
4. If using TLS/mqtts, ensure required TLS options (e.g. `tls_mode`, certs) are provided

Example fix

// before
CREATE SINK s FROM t WITH (connector='mqtt', url='broker:1883');
// after
CREATE SINK s FROM t WITH (connector='mqtt', url='mqtt://broker:1883', qos='at_least_once');

// before CREATE SINK
let url = options.get("url").expect("mqtt url required");
assert!(url.starts_with("mqtt://") || url.starts_with("mqtts://"), "url must include scheme");
let (host, port) = parse_host_port(&url).expect("valid host:port");
tokio::net::TcpStream::connect((host.as_str(), port)).await.expect("broker reachable");

fn is_valid_mqtt_url(u: &str) -> bool {
    matches!(u.split_once("://"), Some(("mqtt" | "mqtts", rest))) && !rest.is_empty()
}

When it happens

Trigger: 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.

Common situations: 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.

Understand the failure class

Background: ECONNREFUSED and "connection refused" / "could not connect to server" errors: what they mean and how to fix them — this error's family across 44 libraries.

Related errors

• Mqtt error: {0}
• Doris/Starrocks connect error: {0}
• SinkError::Nats(anyhow!(e))
• failed to parse response body
• Failed connection {:?},{:?}

AI-assisted analysis of risingwavelabs/risingwave@6469eb736d (2026-09-11). Data as JSON: /api/errors/7e40366f221888aa. Report an issue: GitHub.