risingwavelabs/risingwave · error · SinkError

Mqtt error: {0}

Error message

Mqtt error: {0}

What it means

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.

Source

Thrown at src/connector/src/sink/mod.rs:1136

        anyhow::Error,
    ),
    #[error("coordinator error: {0}")]
    Coordinator(
        #[source]
        #[backtrace]
        anyhow::Error,
    ),
    #[error("ClickHouse error: {0}")]
    ClickHouse(String),
    #[error("Redis error: {0}")]
    Redis(String),
    #[error("Http error: {0}")]
    Http(
        #[source]
        #[backtrace]
        anyhow::Error,
    ),
    #[error("Mqtt error: {0}")]
    Mqtt(
        #[source]
        #[backtrace]
        anyhow::Error,
    ),
    #[error("Nats error: {0}")]
    Nats(
        #[source]
        #[backtrace]
        anyhow::Error,
    ),
    #[error("Google Pub/Sub error: {0}")]
    GooglePubSub(
        #[source]
        #[backtrace]
        anyhow::Error,
    ),
    #[error("Doris/Starrocks connect error: {0}")]

View on GitHub (pinned to 6469eb736d)

Solutions

1. 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.
2. Check TLS/CA settings in the sink definition against the broker's certificate setup.
3. Verify the client id is unique and the account is authorized for the target topic (broker ACLs).
4. Lower QoS or increase broker queue/session limits if publishes fail under load; check broker logs for disconnect reasons.
5. Read the wrapped rumqttc error in RisingWave logs to distinguish connection vs publish-stage failures.

Example fix

// before: credentials rejected at CONNECT
CREATE SINK mqtt_sink FROM mv WITH (
  connector = 'mqtt',
  url = 'mqtt://broker:1883',
  username = 'rw', password = 'old-pass'
);
// after: rotated credentials supplied
CREATE SINK mqtt_sink FROM mv WITH (
  connector = 'mqtt',
  url = 'mqtt://broker:1883',
  username = 'rw', password = 'new-pass'
);

# Verify broker connectivity and auth before creating the sink
mosquitto_pub -h "$MQTT_HOST" -p "$MQTT_PORT" -u "$MQTT_USER" -P "$MQTT_PASS" -t 'rw/health' -m 'check' && echo OK
# or in Rust
let opts = rumqttc::MqttOptions::new("rw-check", host, port);
// attempt connect with a short timeout before sinking

fn is_valid_mqtt_url(url: &str) -> bool {
    url.starts_with("mqtt://") || url.starts_with("mqtts://") || url.starts_with("ssl://")
}

// Distinguish connect-stage from publish-stage failures
match sink_result {
    Err(SinkError::Mqtt(e)) if is_connect_error(&e) => retry_connect_with_backoff(),
    Err(SinkError::Mqtt(e)) => { log(&e); /* check broker logs, QoS, queue limits */ }
    Ok(v) => process(v),
}

When it happens

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

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

Related errors

• Redis error: {0}
• Http error: {0}
• Nats error: {0}
• SinkError::Mqtt(anyhow!(e))
• failed to parse response body

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