risingwavelabs/risingwave · error

streaming jobs {:?} not found

Error message

streaming jobs {:?} not found

What it means

load_fragment_context looks up StreamingJob rows for every job owning the loaded fragments. If some job IDs are absent from the streaming job table, the missing job IDs are returned in this error. It prevents rendering actors for jobs whose catalog entry has vanished.

Source

Thrown at src/meta/src/controller/scale.rs:512

        .map(|fragment| fragment.job_id)
        .collect();

    if job_ids.is_empty() {
        return Ok(LoadedFragmentContext::default());
    }

    let jobs: HashMap<_, _> = StreamingJob::find()
        .filter(streaming_job::Column::JobId.is_in(job_ids.iter().copied().collect_vec()))
        .all(txn)
        .await?
        .into_iter()
        .map(|job| (job.job_id, job))
        .collect();

    let found_job_ids: HashSet<_> = jobs.keys().copied().collect();
    if found_job_ids.len() != job_ids.len() {
        let missing = job_ids.difference(&found_job_ids).copied().collect_vec();
        return Err(anyhow!("streaming jobs {:?} not found", missing).into());
    }

    build_loaded_context(txn, ensembles, fragment_models, jobs).await
}

/// Async load stage for job-scoped rendering. It collects all no-shuffle ensembles and the
/// metadata required to render actor assignments later with a provided worker set.
pub async fn load_fragment_context_for_jobs<C>(
    txn: &C,
    job_ids: HashSet<JobId>,
) -> MetaResult<LoadedFragmentContext>
where
    C: ConnectionTrait,
{
    if job_ids.is_empty() {
        return Ok(LoadedFragmentContext::default());
    }

View on GitHub (pinned to 6469eb736d)

Solutions

1. Check the listed job IDs in the streaming job catalog; if the jobs were dropped, clean up the orphaned fragment rows.
2. Retry the rescale outside any concurrent DDL window.
3. Restore the missing streaming job rows from a meta backup if the jobs should still exist.
4. Report a metadata-consistency bug if the state recurs without concurrent DDL.

let found: HashSet<_> = StreamingJob::find()
    .filter(streaming_job::Column::JobId.is_in(job_ids.iter().copied().collect_vec()))
    .all(txn).await?.into_iter().map(|j| j.job_id).collect();
let missing: Vec<_> = job_ids.difference(&found).collect();
if !missing.is_empty() { bail!("jobs {missing:?} not found; abort rescale"); }

if let Err(e) = load_fragment_context(txn, ensembles).await {
    if e.to_string().contains("streaming jobs") {
        log::warn!("jobs vanished mid-rescale, aborting: {e}");
    }
    return Err(e);
}

When it happens

Trigger: Fragments exist for a job_id but the streaming_jobs row is missing when load_fragment_context queries it — typically fragments orphaned after job deletion, or the job query filtered out a job in a status excluded by the caller's intent.

Common situations: Meta store inconsistency after failed job cancellation (fragments not cascaded); rescale invoked during concurrent DROP MATERIALIZED VIEW / SINK; schema migration leaving orphaned fragment rows.

Understand the failure class

Background: "Not found" and "does not exist" errors: why "Task not found", "No such folder", and "Can't find" fire when a lookup comes back empty — this error's family across 14 libraries.

Related errors

• fragments {:?} not found
• streaming job {job_id} not found
• cannot find StreamActor of actor {}
• Some streaming jobs already exist in meta, please start with
• job fragments should exist for streaming job

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