risingwavelabs/risingwave · error · MetaError

{message}: in worker node {}, {};

Error message

{message}: in worker node {}, {};

What it means

When merge_node_rpc_errors receives one or more per-worker errors, it folds them into a single string: the original message followed by ` in worker node <id>, <report>;` for each failing worker. This aggregated anyhow error is returned so operators see exactly which worker nodes failed and why, in one report.

Source

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

        let mut errors = errors;
        let max_scored = errors
            .extract_if(.., |(_, e)| {
                error_request_copy::<Score>(e) == Some(max_score)
            })
            .next()
            .unwrap();

        return single_error(max_scored);
    }

    // The errors do not have scores, so simply concatenate them.
    let concat: String = errors
        .into_iter()
        .fold(format!("{message}: "), |mut s, (w, e)| {
            write!(&mut s, " in worker node {}, {};", w, e.as_report()).unwrap();
            s
        });
    anyhow!(concat).into()
}

#[cfg(test)]
mod test_partial_graph_id {
    use crate::barrier::rpc::{from_partial_graph_id, to_partial_graph_id};

    #[test]
    fn test_partial_graph_id_conversion() {
        let database_id = 233.into();
        let job_id = 233.into();
        assert_eq!(
            (database_id, None),
            from_partial_graph_id(to_partial_graph_id(database_id, None))
        );
        assert_eq!(
            (database_id, Some(job_id)),
            from_partial_graph_id(to_partial_graph_id(database_id, Some(job_id)))
        );

View on GitHub (pinned to 6469eb736d)

Solutions

1. Read the worker id in the message and inspect that node's logs for the root cause (the trailing report contains the underlying error).
2. Check connectivity between the meta node and the named worker (network, firewall, port).
3. If the worker is dead, let recovery replace it or remove it from the cluster, then retry the operation.
4. If version-skew after upgrade, complete the rolling upgrade so all workers run the same version.

Example fix

// before: ignoring per-node failures until merge produces an opaque aggregate
node_clients.par_iter().for_each(|c| { let _ = c.flush(msg).await; });
// after: fail fast and log per-node results so the aggregate is actionable
for (id, c) in node_clients {
    if let Err(e) = c.flush(msg).await {
        warn!(worker = %id, error = ?e, "worker flush failed");
        failures.push((id, e));
    }
}

// before dispatch: ensure target workers are healthy
let dead: Vec<_> = workers.iter().filter(|w| !is_healthy(w)).collect();
assert!(dead.is_empty(), "unhealthy workers: {:?}", dead);

fn is_worker_node_error(err: &MetaError) -> bool {
    err.to_string().contains("in worker node ")
}

// parse per-worker reports from the aggregate
let msg = e.to_string();
for part in msg.split(" in worker node ").skip(1) {
    let (id, cause) = part.split_once(',').unwrap();
    warn!(worker = id, cause, "worker rpc failed");
}

When it happens

Trigger: Any barrier/foreground RPC dispatched by the meta node to multiple streaming workers where at least one worker returns an error — the failure is reported as `<message>: in worker node <id>, <detail>;`.

Common situations: A worker node crashed or was killed mid-barrier; network partition between meta and a worker; a worker rejected a barrier due to backpressure or being out of sync; version-skew after rolling upgrade.

Related errors

• anyhow!(message.to_owned())
• Invalid worker: {0}, {1}
• Service unavailable: {0}
• Cancelled: {0}
• backup job status not found: job {}, {}

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