{"record":{"id":"2e4668b384e79b38","repo":"risingwavelabs/risingwave","slug":"cannot-connect-to-kafka-broker","errorCode":null,"errorMessage":"cannot connect to kafka broker ({})","messagePattern":"cannot connect to kafka broker \\((.+?)\\)","errorType":"exception","errorClass":"SinkError::Config","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/kafka.rs","lineNumber":406,"sourceCode":"            self.schema.clone(),\n            self.pk_indices.clone(),\n            self.db_name.clone(),\n            self.sink_from_name.clone(),\n            &self.config.common.topic,\n        )\n        .await?;\n\n        // Try Kafka connection.\n        // There is no such interface for kafka producer to validate a connection\n        // use enumerator to validate broker reachability and existence of topic\n        let check = KafkaSplitEnumerator::new(\n            KafkaProperties::from(self.config.clone()),\n            Arc::new(SourceEnumeratorContext::dummy()),\n        )\n        .await?;\n        if let Err(e) = check.check_reachability().await {\n            return Err(SinkError::Config(\n                anyhow!(\n                    \"cannot connect to kafka broker ({})\",\n                    self.config.connection.brokers,\n                )\n                .context(e),\n            ));\n        }\n        Ok(())\n    }\n\n    fn validate_alter_config(config: &BTreeMap<String, String>) -> Result<()> {\n        KafkaConfig::from_btreemap(config.clone())?;\n        Ok(())\n    }\n}\n\n/// When the `DeliveryFuture` the current `future_delivery_buffer`\n/// is buffering is greater than `queue_buffering_max_messages` * `KAFKA_WRITER_MAX_QUEUE_SIZE_RATIO`,\n/// then enforcing commit once","sourceCodeStart":388,"sourceCodeEnd":424,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/kafka.rs#L388-L424","documentation":"validate() performs a reachability check against the configured Kafka brokers using a Kafka connect checker; if check_reachability() fails, this SinkError::Config wraps the underlying error and reports that the sink's brokers could not be connected to at creation time.","triggerScenarios":"Calling Sink::validate() on a KafkaSink where connecting to config.connection.brokers fails — broker down, wrong host/port, DNS failure, auth/TLS misconfiguration, or network isolation from the compute node.","commonSituations":"Typo in bootstrap server address; Kafka not yet started or wrong port; security protocol/SASL misconfigured; firewalls or k8s NetworkPolicy blocking the stream-compute node; brokers behind a service name only resolvable inside the cluster.","solutions":["Verify the broker address/port is reachable (nc/telnet or kcat -L -b brokers) from the compute node","Check broker logs and DNS/service names; use the in-cluster service address for k8s deployments","Review security settings (security.protocol, SASL/TLS credentials) in the WITH options","Retry creation after confirming the Kafka cluster is healthy"],"exampleFix":"// before\nWITH (connector='kafka', properties.bootstrap.server='localhost:9092')\n// after (in-cluster)\nWITH (connector='kafka', properties.bootstrap.server='kafka-service.kafka-ns:9092')","handlingStrategy":"retry","validationCode":"// before CREATE SINK, check reachability\nnc -zv <broker-host> 9092  # or kcat -L -b <brokers>","typeGuard":null,"tryCatchPattern":"if let Err(e) = check.check_reachability().await {\n    warn!(\"kafka brokers {} unreachable: {e}; retrying\", brokers);\n    return Err(SinkError::Config(anyhow!(\"cannot connect to kafka broker ({})\", brokers).context(e)));\n}","preventionTips":["Verify broker DNS/port from the compute node's network context","Match security.protocol/SASL/TLS settings to the broker's listener config","Use k8s service names, not host-mapped addresses, for in-cluster deployments","Confirm the Kafka cluster is up before creating sinks"],"tags":["kafka","network","connectivity","config"],"backgroundTag":"connection-refused","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"}