risingwavelabs/risingwave · error · BatchError

No worker node found for worker slot id

Error message

No worker node found for worker slot id: {}

What it means

Raised by WorkerNodeManager::get_workers_by_worker_slot_ids: an explicit guard that fires when the requested worker slot id list is empty. An empty list would otherwise make the scheduler fail to place any task and silently hang while waiting to schedule the next stage; this error makes the root cause of the scheduling failure explicit. The offending input is an empty worker_slot_ids slice passed by the scheduler.

Solutions

1. Fix the caller (scheduler/planner) to never produce an empty worker slot list for a stage
2. Check cluster membership: if worker slots are derived from active worker nodes, ensure enough compute nodes are registered and running
3. Verify parallelism configuration so at least one parallel unit/worker slot is allocated per stage
4. If triggered by node failures, wait for the meta service to re-register workers, then retry the query

When it happens

Trigger: Thrown at src/batch/src/worker_manager/worker_node_manager.rs:160 when the library encounters an invalid state.

Common situations: See trigger scenarios.

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

Appendix: source

Thrown at src/batch/src/worker_manager/worker_node_manager.rs:160

        write_guard.serving_fragment_vnode_mapping = serving_mapping;
    }

    /// If worker slot ids is empty, the scheduler may fail to schedule any task and stuck at
    /// schedule next stage. If we do not return error in this case, needs more complex control
    /// logic above. Report in this function makes the schedule root fail reason more clear.
    pub fn get_workers_by_worker_slot_ids(
        &self,
        worker_slot_ids: &[WorkerSlotId],
    ) -> Result<Vec<WorkerNode>> {
        if worker_slot_ids.is_empty() {
            return Err(BatchError::EmptyWorkerNodes);
        }
        let guard = self.inner.read().unwrap();
        let mut workers = Vec::with_capacity(worker_slot_ids.len());
        for worker_slot_id in worker_slot_ids {
            match guard.worker_nodes.get(&worker_slot_id.worker_id()) {
                Some(worker) => workers.push((*worker).clone()),
                None => bail!(
                    "No worker node found for worker slot id: {}",
                    worker_slot_id
                ),
            }
        }

        Ok(workers)
    }

    pub fn get_streaming_fragment_mapping(
        &self,
        fragment_id: &FragmentId,
    ) -> Result<WorkerSlotMapping> {
        let guard = self.inner.read().unwrap();

        let Some(streaming_mapping) = guard.streaming_fragment_vnode_mapping.as_ref() else {
            return Err(BatchError::StreamingVnodeMappingNotInitialized);
        };

View on GitHub (pinned to 6469eb736d)