risingwavelabs/risingwave · error · anyhow::Error

duplicate worker id in plan, prev ->

Error message

duplicate worker id {worker_id} in plan, prev {worker_id} -> {dup_change}

What it means

parse_plan builds a per-worker actor-count diff; inserting a worker id that already appeared in the plan means the plan tries to change the same worker twice, which is ambiguous, so parsing bails with this message.

Solutions

1. Deduplicate worker entries so each worker id appears once per plan
2. Merge the counts for repeated worker ids into a single entry
3. Regenerate the plan from the meta tooling, which produces unique worker ids

Example fix

// before
fragment 100 [1:2, 1:5]
// after
fragment 100 [1:5]

let mut seen = HashSet::new();
for w in worker_ids {
    if !seen.insert(w) {
        return Err(anyhow!("duplicate worker id {}", w));
    }
}

When it happens

Trigger: A plan string listing the same worker id in two worker-change entries within (or the parse sees duplicates across) fragment sections, e.g. '1:2,1:5'.

Common situations: Merging plans from multiple runs without deduplicating workers; hand-editing plans and duplicating a worker entry.

Understand the failure class

Background: Conflicting config options: "cannot be used together" — configuration validation errors across open-source libraries — this error's family across 162 libraries.

Related errors

• reschedule failed
• cannot reschedule jobs
• exactly one of job_id or fragment_id must be provided
• get none metadata in commit response for coordinated sink…
• initial epoch greater than aligned initial epoch

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

Appendix: source

Thrown at src/ctl/src/cmd_impl/meta/reschedule.rs:170

        let captures = regex
            .captures(fragment_reschedule_plan)
            .ok_or_else(|| anyhow!("plan \"{}\" format illegal", fragment_reschedule_plan))?;

        let fragment_id = captures
            .get(1)
            .and_then(|mat| mat.as_str().parse::<u32>().ok())
            .ok_or_else(|| anyhow!("plan \"{}\" does not have a valid fragment id", plan))?;

        let worker_changes: Vec<&str> = captures[2].split(',').collect();

        let mut worker_actor_diff = HashMap::new();
        for worker_change in &worker_changes {
            let (worker_id, count) = worker_change.split(':').collect_tuple::<(_, _)>().unwrap();
            let worker_id = worker_id.parse().unwrap();
            let count = count.parse().unwrap();

            if let Some(dup_change) = worker_actor_diff.insert(worker_id, count) {
                anyhow::bail!(
                    "duplicate worker id {worker_id} in plan, prev {worker_id} -> {dup_change}",
                );
            }
        }

        if !worker_actor_diff.is_empty() {
            reschedules.insert(fragment_id, PbWorkerReschedule { worker_actor_diff });
        }
    }
    Ok(reschedules)
}

pub async fn unregister_workers(
    context: &CtlContext,
    workers: Vec<String>,
    yes: bool,
    ignore_not_found: bool,
    check_fragment_occupied: bool,

View on GitHub (pinned to 6469eb736d)