{"record":{"id":"fbf8ed86faec8290","repo":"risingwavelabs/risingwave","slug":"message-in-worker-node","errorCode":null,"errorMessage":"{message}: in worker node {}, {};","messagePattern":"(.+?): in worker node (.+?), (.+?);","errorType":"error_code","errorClass":"MetaError","httpStatus":null,"severity":"error","filePath":"src/meta/src/barrier/rpc.rs","lineNumber":1534,"sourceCode":"        let mut errors = errors;\n        let max_scored = errors\n            .extract_if(.., |(_, e)| {\n                error_request_copy::<Score>(e) == Some(max_score)\n            })\n            .next()\n            .unwrap();\n\n        return single_error(max_scored);\n    }\n\n    // The errors do not have scores, so simply concatenate them.\n    let concat: String = errors\n        .into_iter()\n        .fold(format!(\"{message}: \"), |mut s, (w, e)| {\n            write!(&mut s, \" in worker node {}, {};\", w, e.as_report()).unwrap();\n            s\n        });\n    anyhow!(concat).into()\n}\n\n#[cfg(test)]\nmod test_partial_graph_id {\n    use crate::barrier::rpc::{from_partial_graph_id, to_partial_graph_id};\n\n    #[test]\n    fn test_partial_graph_id_conversion() {\n        let database_id = 233.into();\n        let job_id = 233.into();\n        assert_eq!(\n            (database_id, None),\n            from_partial_graph_id(to_partial_graph_id(database_id, None))\n        );\n        assert_eq!(\n            (database_id, Some(job_id)),\n            from_partial_graph_id(to_partial_graph_id(database_id, Some(job_id)))\n        );","sourceCodeStart":1516,"sourceCodeEnd":1552,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/barrier/rpc.rs#L1516-L1552","documentation":"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.","triggerScenarios":"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>;`.","commonSituations":"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.","solutions":["Read the worker id in the message and inspect that node's logs for the root cause (the trailing report contains the underlying error).","Check connectivity between the meta node and the named worker (network, firewall, port).","If the worker is dead, let recovery replace it or remove it from the cluster, then retry the operation.","If version-skew after upgrade, complete the rolling upgrade so all workers run the same version."],"exampleFix":"// before: ignoring per-node failures until merge produces an opaque aggregate\nnode_clients.par_iter().for_each(|c| { let _ = c.flush(msg).await; });\n// after: fail fast and log per-node results so the aggregate is actionable\nfor (id, c) in node_clients {\n    if let Err(e) = c.flush(msg).await {\n        warn!(worker = %id, error = ?e, \"worker flush failed\");\n        failures.push((id, e));\n    }\n}","handlingStrategy":"try-catch","validationCode":"// before dispatch: ensure target workers are healthy\nlet dead: Vec<_> = workers.iter().filter(|w| !is_healthy(w)).collect();\nassert!(dead.is_empty(), \"unhealthy workers: {:?}\", dead);","typeGuard":"fn is_worker_node_error(err: &MetaError) -> bool {\n    err.to_string().contains(\"in worker node \")\n}","tryCatchPattern":"// parse per-worker reports from the aggregate\nlet msg = e.to_string();\nfor part in msg.split(\" in worker node \").skip(1) {\n    let (id, cause) = part.split_once(',').unwrap();\n    warn!(worker = id, cause, \"worker rpc failed\");\n}","preventionTips":["Keep worker nodes' versions in lockstep during rolling upgrades.","Watch network health between meta and compute nodes; alert on heartbeat loss.","Log the full aggregated message — the worker ids point at the root cause."],"tags":["rpc","meta","worker-node","error-aggregation"],"backgroundTag":"rpc-error-aggregation","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}