{"record":{"id":"89dbde1dd7f8e317","repo":"risingwavelabs/risingwave","slug":"no-coordinator","errorCode":null,"errorMessage":"no coordinator","messagePattern":"no coordinator","errorType":"exception","errorClass":"SinkError::Coordinator","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/mod.rs","lineNumber":903,"sourceCode":"        Ok(())\n    }\n\n    fn validate_unknown_fields(&self) -> Result<()> {\n        Ok(())\n    }\n\n    async fn validate(&self) -> Result<()>;\n    async fn new_log_sinker(&self, writer_param: SinkWriterParam) -> Result<Self::LogSinker>;\n\n    fn is_coordinated_sink(&self) -> bool {\n        false\n    }\n\n    async fn new_coordinator(\n        &self,\n        _iceberg_compact_stat_sender: Option<UnboundedSender<IcebergSinkCompactionUpdate>>,\n    ) -> Result<SinkCommitCoordinator> {\n        Err(SinkError::Coordinator(anyhow!(\"no coordinator\")))\n    }\n}\n\npub trait SinkLogReader: Send {\n    fn start_from(\n        &mut self,\n        start_offset: Option<u64>,\n    ) -> impl Future<Output = LogStoreResult<()>> + Send + '_;\n    /// Emit the next item.\n    ///\n    /// The implementation should ensure that the future is cancellation safe.\n    fn next_item(\n        &mut self,\n    ) -> impl Future<Output = LogStoreResult<(u64, LogStoreReadItem)>> + Send + '_;\n\n    /// Mark that all items emitted so far have been consumed and it is safe to truncate the log\n    /// from the current offset.\n    fn truncate(&mut self, offset: TruncateOffset) -> LogStoreResult<()>;","sourceCodeStart":885,"sourceCodeEnd":921,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/mod.rs#L885-L921","documentation":"This error is returned by the default `new_coordinator` implementation of the sink commit coordinator trait, indicating that this sink connector does not provide a commit coordinator. It surfaces when a two-phase (decoupled) commit is requested for a sink whose implementation has no coordinator, typically non-Iceberg connectors.","triggerScenarios":"Creating a sink with sink decouple enabled (or commit_checkpoint_interval > 1) on a connector whose `SinkCommitter::new_coordinator` uses the default trait implementation that unconditionally returns `Err(SinkError::Coordinator(anyhow!(\"no coordinator\")))`.","commonSituations":"Enabling sink_decouple for a Kafka/Doris/other sink that only supports integrated commit; misconfigured connector type; driver code attempting two-phase commit after a config validation gap.","solutions":["Disable sink decoupling for this sink: `SET sink_decouple = false` and drop `commit_checkpoint_interval` > 1 from WITH options","Check whether the connector supports a commit coordinator (Iceberg does); switch the sink to a supported connector if two-phase commit is required","If implementing a new sink connector, override `new_coordinator` to return a real SinkCommitCoordinator"],"exampleFix":null,"handlingStrategy":"validation","validationCode":"// verify the connector supports a commit coordinator before enabling decouple\nif !matches!(sink_type, SinkType::Iceberg) && (decouple_enabled || commit_interval > 1) {\n    return Err(\"sink decouple requires a coordinator-capable connector (e.g. iceberg)\");\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Only enable sink_decouple for connectors documented to support it (e.g. Iceberg)","Keep commit_checkpoint_interval unset (=1) for integrated-commit sinks"],"tags":["sink","coordinator","risingwave"],"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"}