{"record":{"id":"783bf3ef182baf4f","repo":"risingwavelabs/risingwave","slug":"snowflake-sink-committer-is-not-initialized","errorCode":null,"errorMessage":"Snowflake sink committer is not initialized.","messagePattern":"Snowflake sink committer is not initialized\\.","errorType":"validation","errorClass":"SinkError::Config","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/snowflake_redshift/snowflake.rs","lineNumber":799,"sourceCode":"        Ok(())\n    }\n\n    async fn commit_schema_change(\n        &mut self,\n        _epoch: u64,\n        schema_change: PbSinkSchemaChange,\n    ) -> Result<()> {\n        use risingwave_pb::stream_plan::sink_schema_change::PbOp as SinkSchemaChangeOp;\n        let schema_change_op = schema_change\n            .op\n            .ok_or_else(|| SinkError::Coordinator(anyhow!(\"Invalid schema change operation\")))?;\n        let SinkSchemaChangeOp::AddColumns(add_columns) = schema_change_op else {\n            return Err(SinkError::Coordinator(anyhow!(\n                \"Only AddColumns schema change is supported for Snowflake sink\"\n            )));\n        };\n        let client = self.client.as_mut().ok_or_else(|| {\n            SinkError::Config(anyhow!(\"Snowflake sink committer is not initialized.\"))\n        })?;\n        client\n            .execute_alter_add_columns(\n                &add_columns\n                    .fields\n                    .into_iter()\n                    .map(|f| {\n                        let dt = DataType::from(f.data_type.unwrap());\n                        Ok((f.name, convert_snowflake_data_type(&dt)?))\n                    })\n                    .collect::<Result<Vec<_>>>()?,\n            )\n            .await\n    }\n}\n\nimpl Drop for SnowflakeSinkCommitter {\n    fn drop(&mut self) {","sourceCodeStart":781,"sourceCodeEnd":817,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/snowflake_redshift/snowflake.rs#L781-L817","documentation":"`commit_schema_change` requires the sink's JDBC `client` (committer) to have been initialized. When `self.client` is `None` — e.g. the sink instance is not the designated committer or initialization failed — it returns this Config error before executing the ALTER.","triggerScenarios":"Calling `commit_schema_change` on a Snowflake sink instance whose `client: Option<JdbcClient>` was never populated (non-committer writer instance, or earlier client initialization failure).","commonSituations":"Schema change routed to a writer instance instead of the committer; JDBC client initialization failure earlier swallowed; S3 mode where the JDBC client lifecycle differs.","solutions":["Ensure schema changes are applied through the sink's committer instance where `client` is initialized","Check earlier logs for JDBC client initialization failures and fix connectivity/config first","Restart/resync the sink so the committer re-initializes its client"],"exampleFix":"// before: calling commit_schema_change on a plain writer\nwriter.commit_schema_change(epoch, change).await?; // client is None\n// after: route through the committer that owns the initialized JDBC client\ncommitter.commit_schema_change(epoch, change).await?;","handlingStrategy":"type-guard","validationCode":"if sink.client().is_none() {\n    return Err(\"snowflake committer not initialized; route schema changes to the committer\");\n}","typeGuard":"fn committer_ready(sink: &SnowflakeSinkCommitter) -> bool { sink.client.is_some() }","tryCatchPattern":null,"preventionTips":["Only invoke commit_schema_change on the designated committer instance","Monitor JDBC client initialization health before schema-change events"],"tags":["snowflake","sink","schema-change","state"],"backgroundTag":"invalid-state-transition","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"}