risingwavelabs/risingwave · error · SinkError

SinkError::Config(anyhow!(e))

Error message

SinkError::Config(anyhow!(e))

What it means

In `update_sink_props_by_sink_id`, the meta node parses the sink's stored definition SQL to rewrite its WITH properties. A parse failure (or a result that is not exactly one statement) is returned as `SinkError::Config`, i.e. a sink configuration error. The stored sink definition is expected to be a valid `CREATE SINK` statement.

Source

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

    }

    pub async fn update_sink_props_by_sink_id(
        &self,
        sink_id: SinkId,
        props: BTreeMap<String, String>,
    ) -> MetaResult<HashMap<String, String>> {
        let inner = self.inner.read().await;
        let txn = inner.db.begin().await?;

        let (sink, _obj) = Sink::find_by_id(sink_id)
            .find_also_related(Object)
            .one(&txn)
            .await?
            .ok_or_else(|| MetaError::catalog_id_not_found(ObjectType::Sink.as_str(), sink_id))?;
        validate_sink_props(&sink, &props)?;
        let definition = sink.definition.clone();
        let [mut stmt]: [_; 1] = Parser::parse_sql(&definition)
            .map_err(|e| SinkError::Config(anyhow!(e)))?
            .try_into()
            .unwrap();
        if let Statement::CreateSink { stmt } = &mut stmt {
            update_stmt_with_props(&mut stmt.with_properties.0, &props)?;
        } else {
            panic!("definition is not a create sink statement")
        }
        let mut new_config = sink.properties.clone().into_inner();
        new_config.extend(props.clone());

        let definition = stmt.to_string();
        let active_sink = sink::ActiveModel {
            sink_id: Set(sink_id),
            properties: Set(risingwave_meta_model::Property(new_config.clone())),
            definition: Set(definition),
            ..Default::default()
        };
        Sink::update(active_sink).exec(&txn).await?;

View on GitHub (pinned to 6469eb736d)

Solutions

1. Inspect the stored sink definition (`SHOW SINK <name>`) and check it is a syntactically valid single CREATE SINK statement.
2. Recreate the sink: DROP SINK then CREATE SINK with the desired properties, then re-run the property alteration.
3. Verify the meta store was not hand-edited; restore the original definition if it was modified.
4. If the properties you want are already correct at creation, avoid the alter path by defining them in the original CREATE SINK.

Example fix

-- before: ALTER SINK s ALTER CONNECTOR PROPS ... fails parsing stored definition
-- after
DROP SINK s;
CREATE SINK s FROM mv WITH (connector = 'kafka', properties.add = 'x');

-- before altering sink props, verify the definition parses
SHOW SINK my_sink; -- must be a single valid CREATE SINK statement

match client.alter_sink_props(sink_id, props).await {
    Err(e) if e.to_string().contains("Sink config error") => {
        // fall back to DROP SINK + CREATE SINK with the new props
    }
    other => other?,
}

When it happens

Trigger: Altering sink properties (ALTER SINK ... ALTER CONNECTOR PROPS / update_sink_props_by_sink_id) when the sink's `definition` column cannot be parsed into exactly one statement, or fails to parse as `Statement::CreateSink` (leading to the subsequent panic path).

Common situations: Corrupted or manually edited catalog definitions; definitions produced by older RisingWave versions that the current parser rejects; partial writes after a failed migration.

Understand the failure class

Background: "query failed", "%w: SQL error" — wrapped database query errors in Go libraries explained — this error's family across 3 libraries.

Related errors

• replacement sink catalog requires a sink job
• object {} is not a sink
• Failed to parse source definition SQL
• no state table id in sink: {}
• missing FORMAT ... ENCODE ...

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