{"record":{"id":"b6059e847af41452","repo":"risingwavelabs/risingwave","slug":"coordinator-error-0","errorCode":null,"errorMessage":"coordinator error: {0}","messagePattern":"coordinator error: (.+?)","errorType":"exception","errorClass":"SinkError","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/mod.rs","lineNumber":1120,"sourceCode":"        anyhow::Error,\n    ),\n    #[error(\"Encode error: {0}\")]\n    Encode(String),\n    #[error(\"Avro error: {0}\")]\n    Avro(#[from] apache_avro::Error),\n    #[error(\"Iceberg error: {0}\")]\n    Iceberg(\n        #[source]\n        #[backtrace]\n        anyhow::Error,\n    ),\n    #[error(\"config error: {0}\")]\n    Config(\n        #[source]\n        #[backtrace]\n        anyhow::Error,\n    ),\n    #[error(\"coordinator error: {0}\")]\n    Coordinator(\n        #[source]\n        #[backtrace]\n        anyhow::Error,\n    ),\n    #[error(\"ClickHouse error: {0}\")]\n    ClickHouse(String),\n    #[error(\"Redis error: {0}\")]\n    Redis(String),\n    #[error(\"Http error: {0}\")]\n    Http(\n        #[source]\n        #[backtrace]\n        anyhow::Error,\n    ),\n    #[error(\"Mqtt error: {0}\")]\n    Mqtt(\n        #[source]","sourceCodeStart":1102,"sourceCodeEnd":1138,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/mod.rs#L1102-L1138","documentation":"`SinkError::Coordinator` wraps an `anyhow::Error` and represents failures from sink components that coordinate distributed writes — e.g. Kafka sink's coordinator machinery (transaction/epoch handling, split assignment, metadata operations). Displayed as \"coordinator error: {0}\" with cause and backtrace preserved.","triggerScenarios":"During sink startup or streaming, when coordinator-level operations fail: negotiating with the external system's coordinator/transaction APIs (e.g. Kafka transactions/producer epoch bumping), building the coordinator, or inter-actor coordination RPCs that return errors converted into this variant.","commonSituations":"Kafka broker side effects: `ProducerFenced`/invalid producer epoch after a sink restart or duplicate `transactional.id`; broker unreachable or unsupported transactional APIs; metastore failures for sinks that coordinate via external catalogs; version incompatibilities between client and broker.","solutions":["Read the wrapped message/cause chain to identify which coordination call failed (e.g. InitProducerId, SendOffsetsToTransaction).","If fenced (`ProducerFenced`/`InvalidProducerEpoch`), stop the stale sink instance and restart it so a fresh transactional session is negotiated.","Verify broker reachability and that the broker version supports the transactions/APIs the sink coordinator uses.","Check for duplicate sink writers using the same transactional identifier and remove the duplicate."],"exampleFix":null,"handlingStrategy":"retry","validationCode":"async fn check_transaction_support(admin: &ClusterAdminClient) -> Result<(), String> {\n    let brokers = admin.describe_cluster().await.map_err(|e| e.to_string())?;\n    // ensure at least one broker reachable and version >= 0.11 for transactions\n    anyhow::ensure!(!brokers.is_empty(), \"no kafka brokers reachable\");\n    Ok(())\n}","typeGuard":"fn as_coordinator_error(err: &SinkError) -> Option<&anyhow::Error> {\n    if let SinkError::Coordinator(e) = err { Some(e) } else { None }\n}","tryCatchPattern":"match sink_coordinator.init().await {\n    Err(SinkError::Coordinator(e)) if is_fenced(&e) => {\n        // ProducerFenced / InvalidProducerEpoch: restart with fresh epoch, do not hot-retry\n        log::warn!(\"sink coordinator fenced, re-initializing: {e:#}\");\n        restart_sink_with_new_epoch().await?;\n    }\n    Err(SinkError::Coordinator(e)) => {\n        log::error!(\"coordinator failure: {e:#}\");\n        return Err(e.into());\n    }\n    Ok(_) => {}\n}","preventionTips":["Ensure only one sink writer uses a given transactional.id at a time.","Check broker reachability and version support for transactional APIs before enabling exactly-once.","Use bounded retry with backoff for transient coordinator RPC failures; restart for fencing errors.","Inspect the anyhow source chain with `{e:#}` to distinguish fencing vs network causes."],"tags":["coordinator","distributed","sink","risingwave"],"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"}