risingwavelabs/risingwave · error · anyhow::Error

unconnected worker node {}

Error message

unconnected worker node {}

What it means

`inject_barrier` sends barrier mutations to each worker via its control stream, looked up in `self.workers`. A worker is only usable when its `WorkerNodeState::Connected` holds a control stream; otherwise this error is returned. It means the meta service tried to send a barrier to a worker that is not currently connected (never connected, disconnected, or draining).

Source

Thrown at src/meta/src/barrier/rpc.rs:1298

        let mut node_need_collect = NodeToCollect::new();
        let table_ids_to_sync = table_ids_to_sync.collect_vec();

        node_actors.iter()
            .try_for_each(|(worker_id, actor_ids_to_collect)| {
                assert!(!actor_ids_to_collect.is_empty(), "empty actor_ids_to_collect on worker {worker_id} in node_actors {node_actors:?}");
                let table_ids_to_sync = if nodes_to_sync_table.contains(worker_id) {
                    table_ids_to_sync.clone()
                } else {
                    vec![]
                };

                let node = if let Some((_, worker_state)) = self.workers.get(worker_id)
                    &&
                    let WorkerNodeState::Connected { control_stream, .. } = worker_state
                {
                    control_stream
                } else {
                    return Err(anyhow!("unconnected worker node {}", worker_id).into());
                };

                {
                    let mutation = mutation.clone();
                    let barrier = Barrier {
                        epoch: Some(risingwave_pb::data::Epoch {
                            curr: barrier_info.curr_epoch(),
                            prev: barrier_info.prev_epoch(),
                        }),
                        mutation: mutation.clone().map(|_| BarrierMutation { mutation }),
                        tracing_context: TracingContext::from_span(barrier_info.curr_epoch.span())
                            .to_protobuf(),
                        kind: barrier_info.kind.to_protobuf() as i32,
                    };

                    node.handle
                        .request_sender
                        .send(StreamingControlStreamRequest {

View on GitHub (pinned to 6469eb736d)

Solutions

1. Check the worker's liveness: it likely crashed or lost its gRPC stream; restart the worker node.
2. Verify network connectivity between meta node and the worker (host/port in worker host field).
3. Let barrier recovery run — the failed barrier will be retried once workers reconnect.
4. If a stale worker id is targeted, remove the stale worker registration from the cluster.

// Check worker connectivity before injecting barriers
async fn worker_connected(workers: &HashMap<WorkerId, WorkerNodeState>, id: WorkerId) -> bool {
    matches!(
        workers.get(&id),
        Some(WorkerNodeState::Connected { .. })
    )
}

// Rust: narrow to the Connected variant
fn control_stream_of(state: &WorkerNodeState) -> Option<&ControlStreamHandle> {
    if let WorkerNodeState::Connected { control_stream, .. } = state {
        Some(control_stream)
    } else {
        None
    }
}

match inject_barrier(mutation).await {
    Err(e) if e.to_string().contains("unconnected worker node") => {
        // wait for reconnection, then rely on barrier retry
        tokio::time::sleep(Duration::from_secs(1)).await;
    }
    other => other?,
}

When it happens

Trigger: `inject_barrier` addressing worker_id whose entry in `self.workers` is absent, or whose state is not `WorkerNodeState::Connected` (e.g. disconnected during the barrier).

Common situations: Compute/foreground node crash or network partition mid-barrier; worker just registered but stream not yet established; cluster scale-in removing a node while barriers are in flight; stale worker ids referenced during recovery.

Understand the failure class

Background: 'Could not be found', 'does not exist', 'not found in database': the resource-not-found family when an ID, slug, key, or URI lookup comes back empty — this error's family across 20 libraries.

Related errors

• Doris/Starrocks connect error: {0}
• SinkError::Mqtt(anyhow!(e))
• SinkError::Nats(anyhow!(e))
• failed to send request to {} {:?}
• failed to connect to all meta servers

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