{"record":{"id":"b940132d377b6d0e","repo":"risingwavelabs/risingwave","slug":"job-fragments-should-exist-for-streaming-job","errorCode":null,"errorMessage":"job fragments should exist for streaming job","messagePattern":"job fragments should exist for streaming job","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"critical","filePath":"src/meta/src/controller/scale.rs","lineNumber":369,"sourceCode":"        for (job_id, database_id) in streaming_job_databases {\n            let context = contexts.entry(database_id).or_insert_with(|| {\n                let database_model = database_map\n                    .remove(&database_id)\n                    .expect(\"database should exist for streaming job\");\n                Self {\n                    ensembles: Vec::new(),\n                    job_fragments: HashMap::new(),\n                    job_map: HashMap::new(),\n                    streaming_job_databases: HashMap::new(),\n                    database_map: HashMap::from([(database_id, database_model)]),\n                    fragment_source_ids: HashMap::new(),\n                    fragment_splits: HashMap::new(),\n                }\n            });\n\n            let fragments = job_fragments\n                .remove(&job_id)\n                .expect(\"job fragments should exist for streaming job\");\n            for fragment_id in fragments.keys().copied() {\n                fragment_databases.insert(fragment_id, database_id);\n                if let Some(source_id) = fragment_source_ids.remove(&fragment_id) {\n                    context.fragment_source_ids.insert(fragment_id, source_id);\n                }\n                if let Some(splits) = fragment_splits.remove(&fragment_id) {\n                    context.fragment_splits.insert(fragment_id, splits);\n                }\n            }\n\n            assert!(\n                context\n                    .job_map\n                    .insert(\n                        job_id,\n                        job_map\n                            .remove(&job_id)\n                            .expect(\"streaming job should exist for loaded context\"),","sourceCodeStart":351,"sourceCodeEnd":387,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/controller/scale.rs#L351-L387","documentation":"During CatalogController::into_database_contexts, the loaded fragment context is split per database. For every (job_id, database_id) pair in streaming_job_databases, job_fragments.remove(&job_id) is expected to yield the fragment map. Hitting this expect() panic means a streaming job is recorded in the job-to-database mapping but has no fragment rows in the LoadedFragmentContext built by the load stage.","triggerScenarios":"Calling into_database_contexts on a LoadedFragmentContext where build_loaded_context loaded a job whose streaming_job_databases entry exists but whose fragments were never inserted into job_fragments — e.g. the job's fragments were deleted concurrently, or the load stage assembled job_fragments incompletely relative to streaming_job_databases.","commonSituations":"Metadata inconsistency after a partially-failed scale/replace operation, manual DB edits to the catalog, or a bug in load_fragment_context that skips a job's fragments while still mapping the job to a database during rescale or recovery-driven rescheduling.","solutions":["Inspect the job_id via the panic backtrace and check the meta catalog tables (fragments, streaming jobs) for that job; restore missing fragment rows from a backup or drop/recreate the job.","Verify the load stage (load_fragment_context / build_loaded_context) actually inserts every job in streaming_job_databases into job_fragments; fix the loader if a path skips fragments.","Check for concurrent job cancellation running during rescale; retry the rescale when no DDL is in flight.","If reproducible, file a bug with the meta store snapshot — this is an internal invariant violation."],"exampleFix":"// before\nlet fragments = job_fragments\n    .remove(&job_id)\n    .expect(\"job fragments should exist for streaming job\");\n// after\nlet fragments = job_fragments.remove(&job_id).ok_or_else(|| {\n    anyhow!(\"job fragments for streaming job {job_id} not found in loaded context\")\n})?;","handlingStrategy":"try-catch","validationCode":"// before calling into_database_contexts\nfor (job_id, _db) in &context.streaming_job_databases {\n    assert!(context.job_fragments.contains_key(job_id), \"job {job_id} lacks fragments\");\n}","typeGuard":"fn has_fragments(ctx: &LoadedFragmentContext, job_id: JobId) -> bool { ctx.job_fragments.contains_key(&job_id) }","tryCatchPattern":"// Rust: panics cannot be caught; guard instead.\nmatch context.job_fragments.get(&job_id) {\n    Some(_) => split_ok,\n    None => return Err(anyhow!(\"job fragments missing for {job_id} in loaded context\")),\n}","preventionTips":["Never run rescale concurrently with job DDL","Always build contexts via the standard load_* entry points","Validate LoadedFragmentContext completeness before splitting","Restore meta store from backup rather than hand-editing"],"tags":["meta","rescale","panic","invariant"],"backgroundTag":"internal-invariant-violation","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"}