risingwavelabs/risingwave · error · SinkError::Config

cannot connect to kafka broker ({})

Error message

cannot connect to kafka broker ({})

What it means

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.

Source

Thrown at src/connector/src/sink/kafka.rs:406

            self.schema.clone(),
            self.pk_indices.clone(),
            self.db_name.clone(),
            self.sink_from_name.clone(),
            &self.config.common.topic,
        )
        .await?;

        // Try Kafka connection.
        // There is no such interface for kafka producer to validate a connection
        // use enumerator to validate broker reachability and existence of topic
        let check = KafkaSplitEnumerator::new(
            KafkaProperties::from(self.config.clone()),
            Arc::new(SourceEnumeratorContext::dummy()),
        )
        .await?;
        if let Err(e) = check.check_reachability().await {
            return Err(SinkError::Config(
                anyhow!(
                    "cannot connect to kafka broker ({})",
                    self.config.connection.brokers,
                )
                .context(e),
            ));
        }
        Ok(())
    }

    fn validate_alter_config(config: &BTreeMap<String, String>) -> Result<()> {
        KafkaConfig::from_btreemap(config.clone())?;
        Ok(())
    }
}

/// When the `DeliveryFuture` the current `future_delivery_buffer`
/// is buffering is greater than `queue_buffering_max_messages` * `KAFKA_WRITER_MAX_QUEUE_SIZE_RATIO`,
/// then enforcing commit once

View on GitHub (pinned to 6469eb736d)

Solutions

1. Verify the broker address/port is reachable (nc/telnet or kcat -L -b brokers) from the compute node
2. Check broker logs and DNS/service names; use the in-cluster service address for k8s deployments
3. Review security settings (security.protocol, SASL/TLS credentials) in the WITH options
4. Retry creation after confirming the Kafka cluster is healthy

Example fix

// before
WITH (connector='kafka', properties.bootstrap.server='localhost:9092')
// after (in-cluster)
WITH (connector='kafka', properties.bootstrap.server='kafka-service.kafka-ns:9092')

// before CREATE SINK, check reachability
nc -zv <broker-host> 9092  # or kcat -L -b <brokers>

if let Err(e) = check.check_reachability().await {
    warn!("kafka brokers {} unreachable: {e}; retrying", brokers);
    return Err(SinkError::Config(anyhow!("cannot connect to kafka broker ({})", brokers).context(e)));
}

When it happens

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

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

Understand the failure class

Background: ECONNREFUSED and "connection refused" / "could not connect to server" errors: what they mean and how to fix them — this error's family across 44 libraries.

Related errors

• Must specify '{}' or '{}'
• schemas.enable is expected to be `true` or `false`, got {s}
• serde (de)serialization error for KafkaConfig: {e}
• missing FORMAT ... ENCODE ...
• primary key not defined for {:?} kafka sink (please define i

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