{"record":{"id":"f68a95499a3a080b","repo":"risingwavelabs/risingwave","slug":"object-is-not-a-sink","errorCode":null,"errorMessage":"object {} is not a sink","messagePattern":"object (.+?) is not a sink","errorType":"exception","errorClass":"MetaError","httpStatus":null,"severity":"error","filePath":"src/meta/src/controller/streaming_job.rs","lineNumber":1388,"sourceCode":"                        splits.values().flatten().cloned().collect_vec(),\n                    )\n                })\n                .collect();\n\n            self.update_fragment_splits(&txn, &fragment_splits).await?;\n        }\n\n        if let Some(old_sink_id) = replace_sink {\n            let old_sink_id = *old_sink_id;\n\n            let (old_sink, old_sink_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            if old_sink_object.obj_type != ObjectType::Sink {\n                bail!(\"object {} is not a sink\", old_sink_id);\n            }\n            let final_sink_name = old_sink.name.clone();\n\n            let mut old_objects_to_delete = vec![old_sink_object];\n            old_objects_to_delete\n                .extend(get_belong_objects(&txn, old_sink_id.as_object_id()).await?);\n            let old_object_models = load_object_models(&txn, &old_objects_to_delete).await?;\n            let old_state_table_ids = old_object_models\n                .iter()\n                .filter_map(|object_info| match object_info {\n                    PbObjectInfo::Table(table) => Some(table.id),\n                    _ => None,\n                })\n                .collect_vec();\n            let old_fragment_ids: Vec<FragmentId> = Fragment::find()\n                .select_only()\n                .column(fragment::Column::FragmentId)\n                .filter(","sourceCodeStart":1370,"sourceCodeEnd":1406,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/controller/streaming_job.rs#L1370-L1406","documentation":"During post-collection of fragments for a sink replacement, the object resolved by the given old sink id is not of type Sink. The catalog row exists but belongs to a different relation kind, so the replacement cannot proceed.","triggerScenarios":"post_collect_job_fragments with replace sink flow where old_sink_id points at a table, materialized view, index, or other object type.","commonSituations":"Stale or wrong id passed by the frontend replace flow; user dropped/recreated the sink so the id now maps to another object; id reuse after metadata corruption.","solutions":["Verify the id refers to an actual sink (check rw_catalog object type).","Re-run the replace statement so the frontend resolves the sink id fresh from the catalog.","Recover from the stale catalog state by dropping and recreating the sink."],"exampleFix":null,"handlingStrategy":"validation","validationCode":"-- confirm the id is a sink before replace flow\nSELECT obj_type FROM rw_catalog rw_objects WHERE object_id = <old_sink_id>;","typeGuard":"fn is_sink_object(o: &Object) -> bool { o.obj_type == ObjectType::Sink }","tryCatchPattern":"if let Err(e) = post_replace_flow(..).await {\n    if e.to_string().contains(\"is not a sink\") {\n        // re-resolve the sink id from the catalog by name and retry\n    }\n}","preventionTips":["Pass sink ids resolved fresh from the catalog, never cached ones","Guard against id reuse after drop/recreate"],"tags":["meta","sink","replace","catalog"],"backgroundTag":"type-mismatch","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"}