{"record":{"id":"d0d88f22e3570139","repo":"risingwavelabs/risingwave","slug":"sinkerror-config-anyhow-e-d0d88f","errorCode":null,"errorMessage":"SinkError::Config(anyhow!(e))","messagePattern":"SinkError::Config\\(anyhow!\\(e\\)\\)","errorType":"validation","errorClass":"SinkError","httpStatus":null,"severity":"error","filePath":"src/meta/src/controller/streaming_job.rs","lineNumber":3136,"sourceCode":"    }\n\n    pub async fn update_sink_props_by_sink_id(\n        &self,\n        sink_id: SinkId,\n        props: BTreeMap<String, String>,\n    ) -> MetaResult<HashMap<String, String>> {\n        let inner = self.inner.read().await;\n        let txn = inner.db.begin().await?;\n\n        let (sink, _obj) = Sink::find_by_id(sink_id)\n            .find_also_related(Object)\n            .one(&txn)\n            .await?\n            .ok_or_else(|| MetaError::catalog_id_not_found(ObjectType::Sink.as_str(), sink_id))?;\n        validate_sink_props(&sink, &props)?;\n        let definition = sink.definition.clone();\n        let [mut stmt]: [_; 1] = Parser::parse_sql(&definition)\n            .map_err(|e| SinkError::Config(anyhow!(e)))?\n            .try_into()\n            .unwrap();\n        if let Statement::CreateSink { stmt } = &mut stmt {\n            update_stmt_with_props(&mut stmt.with_properties.0, &props)?;\n        } else {\n            panic!(\"definition is not a create sink statement\")\n        }\n        let mut new_config = sink.properties.clone().into_inner();\n        new_config.extend(props.clone());\n\n        let definition = stmt.to_string();\n        let active_sink = sink::ActiveModel {\n            sink_id: Set(sink_id),\n            properties: Set(risingwave_meta_model::Property(new_config.clone())),\n            definition: Set(definition),\n            ..Default::default()\n        };\n        Sink::update(active_sink).exec(&txn).await?;","sourceCodeStart":3118,"sourceCodeEnd":3154,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/controller/streaming_job.rs#L3118-L3154","documentation":"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.","triggerScenarios":"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).","commonSituations":"Corrupted or manually edited catalog definitions; definitions produced by older RisingWave versions that the current parser rejects; partial writes after a failed migration.","solutions":["Inspect the stored sink definition (`SHOW SINK <name>`) and check it is a syntactically valid single CREATE SINK statement.","Recreate the sink: DROP SINK then CREATE SINK with the desired properties, then re-run the property alteration.","Verify the meta store was not hand-edited; restore the original definition if it was modified.","If the properties you want are already correct at creation, avoid the alter path by defining them in the original CREATE SINK."],"exampleFix":"-- before: ALTER SINK s ALTER CONNECTOR PROPS ... fails parsing stored definition\n-- after\nDROP SINK s;\nCREATE SINK s FROM mv WITH (connector = 'kafka', properties.add = 'x');","handlingStrategy":"try-catch","validationCode":"-- before altering sink props, verify the definition parses\nSHOW SINK my_sink; -- must be a single valid CREATE SINK statement","typeGuard":null,"tryCatchPattern":"match client.alter_sink_props(sink_id, props).await {\n    Err(e) if e.to_string().contains(\"Sink config error\") => {\n        // fall back to DROP SINK + CREATE SINK with the new props\n    }\n    other => other?,\n}","preventionTips":["Do not modify sink definitions directly in the catalog database.","Recreate sinks whose definitions look corrupted rather than altering them.","Keep cluster versions consistent so the parser understands stored definitions."],"tags":["sql","meta","sink","catalog"],"backgroundTag":"sql-query-failed","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"}