{"record":{"id":"485aab13311ae615","repo":"risingwavelabs/risingwave","slug":"job-not-found-in-database","errorCode":null,"errorMessage":"job {} not found in database","messagePattern":"job (.+?) not found in database","errorType":"exception","errorClass":"MetaError","httpStatus":null,"severity":"error","filePath":"src/meta/src/controller/fragment.rs","lineNumber":651,"sourceCode":"        &self,\n        job_id: JobId,\n    ) -> MetaResult<(\n        StreamJobFragments,\n        HashMap<FragmentId, Vec<StreamActor>>,\n        HashMap<ActorId, PbActorStatus>,\n    )> {\n        let inner = self.inner.read().await;\n\n        // Load fragments matching the job from the database\n        let fragments: Vec<_> = FragmentModel::find()\n            .filter(fragment::Column::JobId.eq(job_id))\n            .all(&inner.db)\n            .await?;\n\n        let job_info = StreamingJob::find_by_id(job_id)\n            .one(&inner.db)\n            .await?\n            .ok_or_else(|| anyhow::anyhow!(\"job {} not found in database\", job_id))?;\n\n        let fragment_actors =\n            self.collect_fragment_actor_pairs(fragments, job_info.stream_context())?;\n\n        let job_definition = resolve_streaming_job_definition(&inner.db, &HashSet::from([job_id]))\n            .await?\n            .remove(&job_id);\n\n        Self::compose_table_fragments(\n            job_id,\n            job_info.job_status.into(),\n            job_info.stream_context(),\n            fragment_actors,\n            job_info.max_parallelism as _,\n            job_definition,\n        )\n    }\n","sourceCodeStart":633,"sourceCodeEnd":669,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/controller/fragment.rs#L633-L669","documentation":"get_job_fragments_by_id looks up the streaming job row by primary key in the meta database. If no row exists for `job_id`, it returns an anyhow error 'job {} not found in database', meaning the requested job id does not correspond to any persisted streaming job.","triggerScenarios":"Calling FragmentManager::get_job_fragments_by_id with a job_id that was already dropped, never created, or a table/system id that is not a streaming job row.","commonSituations":"Race between a client fetching job fragments and a DROP MATERIALIZED VIEW completing; stale job id cached in the frontend after the job was removed; using a table id from another environment's meta store.","solutions":["Confirm the job exists: SELECT ... FROM streaming_job WHERE job_id = <id> in the meta DB, or SHOW MATERIALIZED VIEWS","Refresh any cached job id in the client and re-resolve the job by name","Handle the not-found error gracefully on the caller side (e.g. return a 404-equivalent rather than surfacing it as an internal error)","If the job should exist, check meta node logs for a drop job around the failure time"],"exampleFix":"// before\nlet job = StreamingJob::find_by_id(job_id).one(&inner.db).await?.ok_or_else(|| anyhow!(\"job {} not found\", job_id))?;\n// after\nmatch StreamingJob::find_by_id(job_id).one(&inner.db).await? {\n    Some(job) => job,\n    None => return Err(MetaError::catalog(JobNotFound::new(job_id).into())),\n}","handlingStrategy":"validation","validationCode":"let exists = StreamingJob::find_by_id(job_id).one(&db).await?.is_some();\nif !exists { return Err(not_found()); }","typeGuard":null,"tryCatchPattern":"match get_job_fragments_by_id(job_id).await { Ok(f) => f, Err(e) if e.to_string().contains(\"not found in database\") => handle_job_gone(job_id), Err(e) => return Err(e) }","preventionTips":["Re-resolve job ids by name before API calls instead of caching them","Handle job-dropped races explicitly in polling callers","Validate job ids against the catalog before querying"],"tags":["database","job","not-found","meta"],"backgroundTag":"record-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"}