{"record":{"id":"282fae1a9a08f9f6","repo":"risingwavelabs/risingwave","slug":"schema-change-is-not-implemented-for-two-phase-com","errorCode":null,"errorMessage":"Schema change is not implemented for two-phase commit coordinator {}","messagePattern":"Schema change is not implemented for two-phase commit coordinator (.+?)","errorType":"exception","errorClass":"SinkError::Coordinator","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/mod.rs","lineNumber":1005,"sourceCode":"    /// Return serialized commit metadata to be passed to `commit`.\n    async fn pre_commit(\n        &mut self,\n        epoch: u64,\n        metadata: Vec<SinkMetadata>,\n        schema_change: Option<PbSinkSchemaChange>,\n    ) -> Result<Option<Vec<u8>>>;\n\n    /// Idempotent implementation is required, because `commit_data` in the same epoch could be called multiple times.\n    async fn commit_data(&mut self, epoch: u64, commit_metadata: Vec<u8>) -> Result<()>;\n\n    /// Idempotent implementation is required, because `commit_schema_change` in the same epoch could be called multiple\n    /// times.\n    async fn commit_schema_change(\n        &mut self,\n        _epoch: u64,\n        _schema_change: PbSinkSchemaChange,\n    ) -> Result<()> {\n        Err(SinkError::Coordinator(anyhow!(\n            \"Schema change is not implemented for two-phase commit coordinator {}\",\n            std::any::type_name::<Self>()\n        )))\n    }\n\n    /// Idempotent implementation is required, because `abort` in the same epoch could be called multiple times.\n    async fn abort(&mut self, epoch: u64, commit_metadata: Vec<u8>);\n}\n\nimpl SinkImpl {\n    pub fn new(mut param: SinkParam) -> Result<Self> {\n        const PRIVATE_LINK_TARGET_KEY: &str = \"privatelink.targets\";\n\n        // remove privatelink related properties if any\n        param.properties.remove(PRIVATE_LINK_TARGET_KEY);\n\n        let sink_type = param\n            .properties","sourceCodeStart":987,"sourceCodeEnd":1023,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/mod.rs#L987-L1023","documentation":"Identical to the single-phase variant but on the two-phase commit coordinator trait: the default `commit_schema_change` implementation always returns this error, meaning the sink connector does not support committing schema evolution during the two-phase commit protocol.","triggerScenarios":"A `PbSinkSchemaChange` is committed at an epoch against a two-phase commit coordinator (sink decouple enabled) whose type only provides the default `commit_schema_change`, i.e. any connector that has not overridden it (typically triggered by upstream schema alterations).","commonSituations":"Running an ALTER on the source/mv feeding an Iceberg-style decoupled sink with a connector lacking schema evolution; misjudging which connectors support schema-change commit.","solutions":["Defer schema changes until the sink is recreated with the new schema","Switch to a sink connector that implements schema-change commit in its two-phase coordinator","Implement `commit_schema_change` in the connector's coordinator if you own the connector"],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"match err {\n    SinkError::Coordinator(e) if e.to_string().contains(\"two-phase commit coordinator\") => {\n        // handle: pause sink, recreate with new schema, resume\n    }\n    e => return Err(e.into()),\n}","preventionTips":["Restrict ALTERs on sources feeding decoupled sinks to connectors with schema-change commit implemented","Test schema evolution on a staging sink before applying in production"],"tags":["sink","schema-change","two-phase-commit"],"backgroundTag":"method-not-implemented","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"}