risingwavelabs/risingwave · error · SinkError

coordinator error: {0}

Error message

coordinator error: {0}

What it means

`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.

Source

Thrown at src/connector/src/sink/mod.rs:1120

        anyhow::Error,
    ),
    #[error("Encode error: {0}")]
    Encode(String),
    #[error("Avro error: {0}")]
    Avro(#[from] apache_avro::Error),
    #[error("Iceberg error: {0}")]
    Iceberg(
        #[source]
        #[backtrace]
        anyhow::Error,
    ),
    #[error("config error: {0}")]
    Config(
        #[source]
        #[backtrace]
        anyhow::Error,
    ),
    #[error("coordinator error: {0}")]
    Coordinator(
        #[source]
        #[backtrace]
        anyhow::Error,
    ),
    #[error("ClickHouse error: {0}")]
    ClickHouse(String),
    #[error("Redis error: {0}")]
    Redis(String),
    #[error("Http error: {0}")]
    Http(
        #[source]
        #[backtrace]
        anyhow::Error,
    ),
    #[error("Mqtt error: {0}")]
    Mqtt(
        #[source]

View on GitHub (pinned to 6469eb736d)

Solutions

1. Read the wrapped message/cause chain to identify which coordination call failed (e.g. InitProducerId, SendOffsetsToTransaction).
2. If fenced (`ProducerFenced`/`InvalidProducerEpoch`), stop the stale sink instance and restart it so a fresh transactional session is negotiated.
3. Verify broker reachability and that the broker version supports the transactions/APIs the sink coordinator uses.
4. Check for duplicate sink writers using the same transactional identifier and remove the duplicate.

async fn check_transaction_support(admin: &ClusterAdminClient) -> Result<(), String> {
    let brokers = admin.describe_cluster().await.map_err(|e| e.to_string())?;
    // ensure at least one broker reachable and version >= 0.11 for transactions
    anyhow::ensure!(!brokers.is_empty(), "no kafka brokers reachable");
    Ok(())
}

fn as_coordinator_error(err: &SinkError) -> Option<&anyhow::Error> {
    if let SinkError::Coordinator(e) = err { Some(e) } else { None }
}

match sink_coordinator.init().await {
    Err(SinkError::Coordinator(e)) if is_fenced(&e) => {
        // ProducerFenced / InvalidProducerEpoch: restart with fresh epoch, do not hot-retry
        log::warn!("sink coordinator fenced, re-initializing: {e:#}");
        restart_sink_with_new_epoch().await?;
    }
    Err(SinkError::Coordinator(e)) => {
        log::error!("coordinator failure: {e:#}");
        return Err(e.into());
    }
    Ok(_) => {}
}

When it happens

Trigger: 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.

Common situations: 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.

Understand the failure class

Background: "Invalid state transition" errors: "status must be X, actually Y", "already rejected/charging/uninstalled", "cannot ... while running" — what they mean when a library rejects your call — this error's family across 31 libraries.

Related errors

• no coordinator
• should have meta client
• should get metadata on checkpoint barrier
• newly start epoch {} after update vnode bitmap not matched w
• HTTP sink url option cannot coexist with url column

AI-assisted analysis of risingwavelabs/risingwave@6469eb736d (2026-09-11). Data as JSON: /api/errors/b6059e847af41452. Report an issue: GitHub.