{"record":{"id":"1b6e9fe690722dad","repo":"risingwavelabs/risingwave","slug":"old-sink-job-not-found-in-barrier-state","errorCode":null,"errorMessage":"old sink job {} not found in barrier state","messagePattern":"old sink job (.+?) not found in barrier state","errorType":"validation","errorClass":"MetaError","httpStatus":null,"severity":"error","filePath":"src/meta/src/barrier/checkpoint/state.rs","lineNumber":954,"sourceCode":"                        }),\n                );\n                if let CreateStreamingJobType::SinkIntoTable(ref ctx) = job_type {\n                    let downstream_fragment_id = ctx.new_sink_downstream.downstream_fragment_id;\n                    self.database_info.pre_apply_add_node_upstream(\n                        downstream_fragment_id,\n                        &PbUpstreamSinkInfo {\n                            upstream_fragment_id: ctx.sink_fragment_id,\n                            sink_output_schema: ctx.sink_output_fields.clone(),\n                            project_exprs: ctx.project_exprs.clone(),\n                        },\n                    );\n                }\n\n                let (table_ids, node_actors) = self.collect_base_info();\n                let dropped_actors = if let Some(old_sink_job_id) = old_sink_job_id {\n                    let Some(job) = self.database_info.post_apply_remove_job(old_sink_job_id)\n                    else {\n                        bail!(\n                            \"old sink job {} not found in barrier state\",\n                            old_sink_job_id\n                        );\n                    };\n                    job.fragment_infos\n                        .values()\n                        .flat_map(|fragment| fragment.actors.keys().copied())\n                        .collect()\n                } else {\n                    vec![]\n                };\n\n                // Actors to create\n                let actors_to_create = Some(Command::create_streaming_job_actors_to_create(\n                    &info,\n                    &mut edges,\n                    &actors.stream_actors,\n                    &actors.actor_location,","sourceCodeStart":936,"sourceCodeEnd":972,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/barrier/checkpoint/state.rs#L936-L972","documentation":"During a replace-sink apply, the old sink job id being replaced was not found in the barrier state's job registry (post_apply_remove_job returned None), so the applier cannot collect its dropped actors and bails out.","triggerScenarios":"Applying a replace-sink command via handle_new_barrier where old_sink_job_id references a sink that no longer exists in database_info (already dropped, finished, or failed).","commonSituations":"Concurrent DROP SINK and replace-sink; retry of a replace after a failure that already removed the old job; stale metadata after recovery.","solutions":["Verify the old sink still exists before issuing the replace","Retry the operation; if the sink is gone, create a new sink instead","Check for concurrent DDL races and re-run after the conflicting command completes","Report a bug if it reproduces on a stable barrier state"],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":"if barrier_state.database_info.get_job(old_sink_job_id).is_none() {\n    return Err(\"old sink already removed; create new sink instead\");\n}","typeGuard":"fn old_sink_exists(state: &BarrierState, id: JobId) -> bool {\n    state.database_info.get_job(id).map(|j| j.is_sink()).unwrap_or(false)\n}","tryCatchPattern":"match replace_sink(...).await {\n    Err(e) if e.contains(\"not found in barrier state\") => create_sink_fresh().await,\n    other => other,\n}","preventionTips":["Avoid concurrent DROP SINK and replace-sink on the same sink","Confirm sink existence before issuing replace","Serialize DDL operations on the same sink"],"tags":["meta","barrier","sink","record-not-found"],"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"}