risingwavelabs/risingwave · error

Sink not found: {}

Error message

Sink not found: {}

What it means

get_sink_param loads the sink catalog row by id via the catalog controller. When no sink exists with the given id, it fails with 'Sink not found: {id}' before Iceberg compaction configuration can be loaded for it.

Source

Thrown at src/meta/src/manager/iceberg_compaction/mod.rs:116

                    manifest_rewrite_sink_ids: HashSet::default(),
                    manual_compaction_waiters: HashMap::default(),
                })),
                metadata_manager,
                iceberg_compactor_manager,
                compactor_streams_change_tx,
                metrics,
            }),
            compactor_streams_change_rx,
        )
    }

    async fn get_sink_param(&self, sink_id: SinkId) -> MetaResult<SinkParam> {
        let prost_sink_catalog = self
            .metadata_manager
            .catalog_controller
            .get_sink_by_id(sink_id)
            .await?
            .ok_or_else(|| anyhow!("Sink not found: {}", sink_id))?;
        let sink_catalog = SinkCatalog::from(prost_sink_catalog);
        let param = SinkParam::try_from_sink_catalog(sink_catalog)?;
        Ok(param)
    }

    async fn load_iceberg_config(&self, sink_id: SinkId) -> MetaResult<IcebergConfig> {
        let sink_param = self.get_sink_param(sink_id).await?;
        let iceberg_config = IcebergConfig::from_btreemap(sink_param.properties)?;
        Ok(iceberg_config)
    }

    /// Clear the iceberg maintenance state of the sink aborted by
    /// `try_abort_creating_streaming_job`, if any.
    pub fn clear_maintenance_for_aborted_job(&self, abort_result: &AbortCreatingJobResult) {
        for &sink_id in &abort_result.aborted_sink_ids {
            self.clear_iceberg_maintenance_by_sink_id(sink_id);
        }
    }

View on GitHub (pinned to 6469eb736d)

Solutions

1. Confirm the sink exists: SELECT ... FROM rw_catalog.rw_sinks WHERE sink_id = <id>.
2. If the sink was dropped, ignore/清理 the stale queue entry; GC should skip missing sinks.
3. If calling manually, use the correct sink id (not table id or sink name).
4. Check for DROP SINK racing with scheduled compaction; re-trigger after catalog settles.

Example fix

// before: wrong id passed (name instead of id)
CALL rw_iceberg_compaction('my_iceberg_sink');
// after: resolve id first
SELECT sink_id FROM rw_catalog.rw_sinks WHERE name = 'my_iceberg_sink';
CALL rw_iceberg_compaction(<sink_id>);

-- SQL: resolve and verify the sink id first
SELECT sink_id FROM rw_catalog.rw_sinks WHERE sink_id = $1 OR name = $2;

match load_iceberg_config(sink_id).await {
    Err(e) if e.to_string().starts_with("Sink not found") => {
        // stale id: refresh sink registry and skip this sink in the batch
        skip_sink(sink_id);
    }
    other => other?,
}

When it happens

Trigger: load_iceberg_config -> get_sink_param(sink_id) where get_sink_by_id returns None — the sink was dropped, never existed, or the id is stale (e.g. referenced from a leftover compaction/gc queue).

Common situations: Sink dropped while background GC/compaction still holds its id; typo'd sink id in a manual CALL; meta catalog restored from a backup lacking the sink; concurrent DROP SINK racing with a scheduled compaction.

Understand the failure class

Background: Record Not Found Errors: "not found", RecordNotFound, and "was not found" — what they mean and how to fix them — this error's family across 28 libraries.

Related errors

• {e}
• failed to create iceberg table
• Iceberg branch {} disappeared after manifest rewrite for sin
• Manual iceberg compaction task {} for sink {} failed: {}
• No iceberg compactor available

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