risingwavelabs/risingwave · error · MetaError

object {} is not a sink

Error message

object {} is not a sink

What it means

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.

Source

Thrown at src/meta/src/controller/streaming_job.rs:1388

                        splits.values().flatten().cloned().collect_vec(),
                    )
                })
                .collect();

            self.update_fragment_splits(&txn, &fragment_splits).await?;
        }

        if let Some(old_sink_id) = replace_sink {
            let old_sink_id = *old_sink_id;

            let (old_sink, old_sink_object) = Sink::find_by_id(old_sink_id)
                .find_also_related(Object)
                .one(&txn)
                .await?
                .and_then(|(sink, object)| object.map(|object| (sink, object)))
                .ok_or_else(|| MetaError::catalog_id_not_found("sink", old_sink_id))?;
            if old_sink_object.obj_type != ObjectType::Sink {
                bail!("object {} is not a sink", old_sink_id);
            }
            let final_sink_name = old_sink.name.clone();

            let mut old_objects_to_delete = vec![old_sink_object];
            old_objects_to_delete
                .extend(get_belong_objects(&txn, old_sink_id.as_object_id()).await?);
            let old_object_models = load_object_models(&txn, &old_objects_to_delete).await?;
            let old_state_table_ids = old_object_models
                .iter()
                .filter_map(|object_info| match object_info {
                    PbObjectInfo::Table(table) => Some(table.id),
                    _ => None,
                })
                .collect_vec();
            let old_fragment_ids: Vec<FragmentId> = Fragment::find()
                .select_only()
                .column(fragment::Column::FragmentId)
                .filter(

View on GitHub (pinned to 6469eb736d)

Solutions

1. Verify the id refers to an actual sink (check rw_catalog object type).
2. Re-run the replace statement so the frontend resolves the sink id fresh from the catalog.
3. Recover from the stale catalog state by dropping and recreating the sink.

-- confirm the id is a sink before replace flow
SELECT obj_type FROM rw_catalog rw_objects WHERE object_id = <old_sink_id>;

fn is_sink_object(o: &Object) -> bool { o.obj_type == ObjectType::Sink }

if let Err(e) = post_replace_flow(..).await {
    if e.to_string().contains("is not a sink") {
        // re-resolve the sink id from the catalog by name and retry
    }
}

When it happens

Trigger: post_collect_job_fragments with replace sink flow where old_sink_id points at a table, materialized view, index, or other object type.

Common situations: 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.

Understand the failure class

Background: Type mismatch errors: IllegalArgumentException, TypeError and type guards across 150 open-source libraries — this error's family across 150 libraries.

Related errors

• replacement sink catalog requires a sink job
• old sink {} does not match replacement sink {}
• replace sink into table is not supported
• sink {} is not ready to be replaced
• SinkError::Config(anyhow!(e))

AI-assisted analysis of risingwavelabs/risingwave@6469eb736d (2026-09-11). Data as JSON: /api/errors/f68a95499a3a080b. Report an issue: GitHub.