{"record":{"id":"e8d128de128ae818","repo":"risingwavelabs/risingwave","slug":"replacement-sink-catalog-requires-a-sink-job","errorCode":null,"errorMessage":"replacement sink catalog requires a sink job","messagePattern":"replacement sink catalog requires a sink job","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/meta/src/controller/streaming_job.rs","lineNumber":504,"sourceCode":"            (None, DefaultParallelism::Full) => StreamingParallelism::Adaptive,\n            (None, DefaultParallelism::Default(n)) => StreamingParallelism::Fixed(n.get()),\n            (Some(n), _) => StreamingParallelism::Fixed(n.parallelism as _),\n        };\n        let backfill_parallelism = backfill_parallelism\n            .as_ref()\n            .map(|p| StreamingParallelism::Fixed(p.parallelism as _))\n            .or_else(|| {\n                backfill_adaptive_parallelism_strategy\n                    .as_ref()\n                    .map(|_| StreamingParallelism::Adaptive)\n            });\n\n        ensure_user_id(streaming_job.owner() as _, &txn).await?;\n        ensure_object_id(ObjectType::Database, streaming_job.database_id(), &txn).await?;\n        ensure_object_id(ObjectType::Schema, streaming_job.schema_id(), &txn).await?;\n        if let Some(old_sink_id) = replace_sink {\n            let StreamingJob::Sink(sink, _) = streaming_job else {\n                bail!(\"replacement sink catalog requires a sink job\")\n            };\n            let (old_sink, old_object) = Sink::find_by_id(*old_sink_id)\n                .find_also_related(Object)\n                .one(&txn)\n                .await?\n                .and_then(|(sink, object)| object.map(|object| (sink, object)))\n                .ok_or_else(|| MetaError::catalog_id_not_found(\"sink\", *old_sink_id))?;\n            let old_streaming_job = StreamingJobModel::find_by_id(old_sink_id.as_job_id())\n                .one(&txn)\n                .await?\n                .ok_or_else(|| MetaError::catalog_id_not_found(\"sink\", *old_sink_id))?;\n            if old_object.obj_type != ObjectType::Sink\n                || old_object.database_id != Some(sink.database_id)\n                || old_object.schema_id != Some(sink.schema_id)\n                || old_sink.name != sink.name\n            {\n                bail!(\n                    \"old sink {} does not match replacement sink {}\",","sourceCodeStart":486,"sourceCodeEnd":522,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/controller/streaming_job.rs#L486-L522","documentation":"A CREATE SINK statement was issued with a replace target (replacing an existing sink catalog), but the new job being created is not a sink job. Replacement of a sink catalog is only valid when the replacement itself is a sink.","triggerScenarios":"Invoking create_job_catalog with replace_sink set to Some(old_sink_id) while the streaming_job is a Mv/Table/MaterializedView variant instead of StreamingJob::Sink.","commonSituations":"Frontend planner bug or misuse of the replace-sink internal API where an `ALTER`/replace flow substitutes a non-sink relation.","solutions":["Ensure the replacement job is a sink (CREATE SINK ... REPLACE syntax path).","Check the frontend planner that resolves replacement targets is passing the correct job variant.","Remove the replace option if a non-sink object was intended and create it as a new job."],"exampleFix":null,"handlingStrategy":"validation","validationCode":"// frontend-side check before issuing replace\nif !matches!(streaming_job, StreamingJob::Sink(..)) {\n    // do not set replace_sink\n}","typeGuard":"fn is_sink_job(job: &StreamingJob) -> bool { matches!(job, StreamingJob::Sink(..)) }","tryCatchPattern":"match catalog.create_job_catalog(job).await {\n    Err(e) if e.to_string().contains(\"requires a sink job\") => /* clear replace option or send a sink job */,\n    other => other?,\n}","preventionTips":["Only set the replace option on the sink-replacement code path","Add a frontend assertion that replaced jobs are sinks"],"tags":["meta","sink","catalog"],"backgroundTag":"invalid-argument-value","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"}