{"record":{"id":"cb5d027aee54c3c9","repo":"risingwavelabs/risingwave","slug":"streaming-jobs-not-found","errorCode":null,"errorMessage":"streaming jobs {:?} not found","messagePattern":"streaming jobs (.+?) not found","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/meta/src/controller/scale.rs","lineNumber":512,"sourceCode":"        .map(|fragment| fragment.job_id)\n        .collect();\n\n    if job_ids.is_empty() {\n        return Ok(LoadedFragmentContext::default());\n    }\n\n    let jobs: HashMap<_, _> = StreamingJob::find()\n        .filter(streaming_job::Column::JobId.is_in(job_ids.iter().copied().collect_vec()))\n        .all(txn)\n        .await?\n        .into_iter()\n        .map(|job| (job.job_id, job))\n        .collect();\n\n    let found_job_ids: HashSet<_> = jobs.keys().copied().collect();\n    if found_job_ids.len() != job_ids.len() {\n        let missing = job_ids.difference(&found_job_ids).copied().collect_vec();\n        return Err(anyhow!(\"streaming jobs {:?} not found\", missing).into());\n    }\n\n    build_loaded_context(txn, ensembles, fragment_models, jobs).await\n}\n\n/// Async load stage for job-scoped rendering. It collects all no-shuffle ensembles and the\n/// metadata required to render actor assignments later with a provided worker set.\npub async fn load_fragment_context_for_jobs<C>(\n    txn: &C,\n    job_ids: HashSet<JobId>,\n) -> MetaResult<LoadedFragmentContext>\nwhere\n    C: ConnectionTrait,\n{\n    if job_ids.is_empty() {\n        return Ok(LoadedFragmentContext::default());\n    }\n","sourceCodeStart":494,"sourceCodeEnd":530,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/controller/scale.rs#L494-L530","documentation":"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.","triggerScenarios":"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.","commonSituations":"Meta store inconsistency after failed job cancellation (fragments not cascaded); rescale invoked during concurrent DROP MATERIALIZED VIEW / SINK; schema migration leaving orphaned fragment rows.","solutions":["Check the listed job IDs in the streaming job catalog; if the jobs were dropped, clean up the orphaned fragment rows.","Retry the rescale outside any concurrent DDL window.","Restore the missing streaming job rows from a meta backup if the jobs should still exist.","Report a metadata-consistency bug if the state recurs without concurrent DDL."],"exampleFix":null,"handlingStrategy":"validation","validationCode":"let found: HashSet<_> = StreamingJob::find()\n    .filter(streaming_job::Column::JobId.is_in(job_ids.iter().copied().collect_vec()))\n    .all(txn).await?.into_iter().map(|j| j.job_id).collect();\nlet missing: Vec<_> = job_ids.difference(&found).collect();\nif !missing.is_empty() { bail!(\"jobs {missing:?} not found; abort rescale\"); }","typeGuard":null,"tryCatchPattern":"if let Err(e) = load_fragment_context(txn, ensembles).await {\n    if e.to_string().contains(\"streaming jobs\") {\n        log::warn!(\"jobs vanished mid-rescale, aborting: {e}\");\n    }\n    return Err(e);\n}","preventionTips":["Check job status before including it in a rescale","Clean up fragments when cancelling jobs (cascade)","Run rescale outside DDL windows","Verify catalog consistency after upgrades"],"tags":["meta","rescale","streaming-job","metadata"],"backgroundTag":"entity-not-found","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"}