{"record":{"id":"559859b53b7a69bc","repo":"risingwavelabs/risingwave","slug":"kafka-error-0","errorCode":null,"errorMessage":"Kafka error: {0}","messagePattern":"Kafka error: (.+?)","errorType":"exception","errorClass":"SinkError","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/mod.rs","lineNumber":1090,"sourceCode":"        }\n\n        $(\n            impl From<$sink_type> for SinkImpl {\n                fn from(sink: $sink_type) -> SinkImpl {\n                    SinkImpl::$variant_name(Box::new(sink))\n                }\n            }\n        )*\n    };\n}\n\ndef_sink_impl!();\n\npub type Result<T> = std::result::Result<T, SinkError>;\n\n#[derive(Error, Debug)]\npub enum SinkError {\n    #[error(\"Kafka error: {0}\")]\n    Kafka(#[from] rdkafka::error::KafkaError),\n    #[error(\"Kinesis error: {0}\")]\n    Kinesis(\n        #[source]\n        #[backtrace]\n        anyhow::Error,\n    ),\n    #[error(\"Remote sink error: {0}\")]\n    Remote(\n        #[source]\n        #[backtrace]\n        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}\")]","sourceCodeStart":1072,"sourceCodeEnd":1108,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/mod.rs#L1072-L1108","documentation":"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: ...'.","triggerScenarios":"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.","commonSituations":"Kafka brokers unreachable due to network/advertised-listener misconfig; wrong topic name (unknown topic); message exceeding broker `message.max.bytes`; auth (SASL/SSL) failures.","solutions":["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","Verify bootstrap servers are reachable: `nc -vz <host> <port>` from the RisingWave host","Confirm topic exists and user has produce/consume ACLs; check `properties.bootstrap.server` and security options in WITH"],"exampleFix":null,"handlingStrategy":"retry","validationCode":"// pre-check broker connectivity before creating the Kafka sink\nnc -vz <broker-host> 9092 || echo \"broker unreachable\"","typeGuard":null,"tryCatchPattern":"match err {\n    SinkError::Kafka(e) if e.to_string().contains(\"Message production timed out\") => retry_with_backoff(),\n    SinkError::Kafka(e) => { log::error!(\"kafka error: {e}\"); return Err(e.into()); }\n    e => return Err(e.into()),\n}","preventionTips":["Verify bootstrap servers, SASL/SSL settings, and topic existence before CREATE SINK","Check broker message.max.bytes vs your row size; monitor broker reachability from compute nodes"],"tags":["kafka","sink","network","rdkafka"],"backgroundTag":"http-request-failed","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T21:17:11.552Z"}