risingwavelabs/risingwave · error · ConnectorError

Nats connection status is not connected, current status is

Error message

Nats connection status is not connected, current status is {:?}

What it means

The NATS source enumerator's list_splits refuses to run when the underlying async_nats client is not in the Connected state. Returning a default split without this check would cause downstream executors to crash, so the connector fails fast with the actual connection state for diagnosis.

Solutions

1. Verify the NATS server is running and reachable at the configured URL (e.g. nats://host:4222) with `nats server report` or by checking the broker process.
2. Check credentials/TLS settings — a failed auth handshake leaves the client in a non-Connected state.
3. Retry the operation after the async_nats client auto-reconnects; the connection status will return to Connected.
4. Inspect logs for the reported state (e.g. Connecting, Reconnecting, Disconnected) to pinpoint the connection problem.

Example fix

// before
bail!("Nats connection status is not connected, current status is {:?}", state);
// after (caller-side retry until connected)
wait_until_connected(&client).await?; // retry loop honoring async_nats reconnect policy
let splits = enumerator.list_splits().await?;

if client.state() != async_nats::connection::State::Connected {
    return Err("nats client not connected; retry after reconnect");
}

match result {
    Err(e) if e.to_string().contains("connection status is not connected") => retry_with_backoff(|| list_splits()).await,
    other => other,
}

When it happens

Trigger: Calling list_splits on a NatsSplitEnumerator whose client.connection_state() is anything other than async_nats::connection::State::Connected — e.g. before the initial connection is established, or after the broker connection dropped and has not reconnected.

Common situations: NATS broker is down or unreachable during source startup; network partition between RisingWave and NATS server; source started before NATS finishes accepting connections; TLS/auth rejected causing the client to sit in a reconnecting state.

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

• connector error
• additional column is not supported for connector
• additional column type
• Connector error
• ConnectorError

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

Appendix: source

Thrown at src/connector/src/source/nats/enumerator/mod.rs:64

        // check if the stream exists or allow create stream
        let jetstream = NatsCommon::build_context_from_client(&client);
        let _ = properties
            .common
            .build_or_get_stream(jetstream, properties.stream.clone())
            .await?;
        Ok(Self {
            subject: properties.common.subject,
            split_id: Arc::from("0"),
            client,
        })
    }

    async fn list_splits(&mut self) -> ConnectorResult<Vec<NatsSplit>> {
        // Nats currently does not support list_splits API, if we simple return the default 0 without checking the client status, will result executor crash
        let state = self.client.connection_state();
        if state != async_nats::connection::State::Connected {
            bail!(
                "Nats connection status is not connected, current status is {:?}",
                state
            );
        }
        // TODO: to simplify the logic, return 1 split for first version
        let nats_split = NatsSplit {
            subject: self.subject.clone(),
            split_id: Arc::from("0"), // be the same as `from_nats_jetstream_message`
            start_sequence: NatsOffset::None,
        };

        Ok(vec![nats_split])
    }
}

View on GitHub (pinned to 6469eb736d)