risingwavelabs/risingwave · error · SinkError

Kafka error: {0}

Error message

Kafka error: {0}

What it means

Variant of `SinkError` wrapping `rdkafka::error::KafkaError`. Any failure surfaced by the rdkafka client library (broker connection failures, message production timeouts, partition errors, consumer errors) is converted via `#[from]` into this variant and rendered as 'Kafka error: ...'.

Source

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

        }

        $(
            impl From<$sink_type> for SinkImpl {
                fn from(sink: $sink_type) -> SinkImpl {
                    SinkImpl::$variant_name(Box::new(sink))
                }
            }
        )*
    };
}

def_sink_impl!();

pub type Result<T> = std::result::Result<T, SinkError>;

#[derive(Error, Debug)]
pub enum SinkError {
    #[error("Kafka error: {0}")]
    Kafka(#[from] rdkafka::error::KafkaError),
    #[error("Kinesis error: {0}")]
    Kinesis(
        #[source]
        #[backtrace]
        anyhow::Error,
    ),
    #[error("Remote sink error: {0}")]
    Remote(
        #[source]
        #[backtrace]
        anyhow::Error,
    ),
    #[error("Encode error: {0}")]
    Encode(String),
    #[error("Avro error: {0}")]
    Avro(#[from] apache_avro::Error),
    #[error("Iceberg error: {0}")]

View on GitHub (pinned to 6469eb736d)

Solutions

1. Check the inner rdkafka message for the root cause (e.g. 'Broker transport failure' vs 'Message size too large') and fix connectivity, SASL/SSL settings, or size limits
2. Verify bootstrap servers are reachable: `nc -vz <host> <port>` from the RisingWave host
3. Confirm topic exists and user has produce/consume ACLs; check `properties.bootstrap.server` and security options in WITH

// pre-check broker connectivity before creating the Kafka sink
nc -vz <broker-host> 9092 || echo "broker unreachable"

match err {
    SinkError::Kafka(e) if e.to_string().contains("Message production timed out") => retry_with_backoff(),
    SinkError::Kafka(e) => { log::error!("kafka error: {e}"); return Err(e.into()); }
    e => return Err(e.into()),
}

When it happens

Trigger: Kafka sink producing messages fails (broker down, message too large, timeouts), or Kafka source/sink consumer hits rdkafka errors during poll/commit; any `rdkafka::KafkaError` propagated through `Result<SinkError>` plumbing.

Common situations: Kafka brokers unreachable due to network/advertised-listener misconfig; wrong topic name (unknown topic); message exceeding broker `message.max.bytes`; auth (SASL/SSL) failures.

Understand the failure class

Background: 'Something went wrong' / 'Request failed (500)' / 'HTTP error! status: 404' — what failed HTTP requests actually mean and how to find the real cause — this error's family across 28 libraries.

Related errors

• Sink error: sink_id={1}, error: {0}
• failed to parse response body
• Failed connection {:?},{:?}
• channel closed
• sending stream load request failed

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