stalwartlabs/stalwart · error

Cluster::SubscriberError

Error message

Cluster::SubscriberError

What it means

The Kafka coordinator pub-sub layer wraps any failure while creating or subscribing a Kafka StreamConsumer in a trc::Error tagged Cluster::SubscriberError. Error 20 is raised at the consumer-creation step: `create_with_context(CustomContext)` failed, so no Kafka subscriber could be built for the requested topic. The underlying librdkafka error is attached via `.reason(err)`.

Source

Thrown at crates/coordinator/src/backend/kafka/pubsub.rs:40

    pub async fn publish(&self, topic: &'static str, message: Vec<u8>) -> trc::Result<()> {
        self.producer
            .send(
                FutureRecord::<(), [u8]>::to(topic).payload(message.as_slice()),
                Duration::from_secs(0),
            )
            .await
            .map(|_| ())
            .map_err(|(err, _)| {
                Error::new(EventType::Cluster(ClusterEvent::PublisherError)).reason(err)
            })
    }

    pub async fn subscribe(&self, topic: &'static str) -> trc::Result<PubSubStream> {
        let subs: StreamConsumer<CustomContext> = self
            .consumer_builder
            .create_with_context(CustomContext)
            .map_err(|err| {
                Error::new(EventType::Cluster(ClusterEvent::SubscriberError)).reason(err)
            })?;
        subs.subscribe(&[topic]).map_err(|err| {
            Error::new(EventType::Cluster(ClusterEvent::SubscriberError)).reason(err)
        })?;

        Ok(PubSubStream::Kafka(KafkaPubSubStream { subs }))
    }
}

impl KafkaPubSubStream {
    pub async fn next(&mut self) -> Option<Msg> {
        let msg = self.subs.recv().await.ok()?;
        let _ = self.subs.commit_message(&msg, CommitMode::Async);
        Msg::Kafka(msg.payload().unwrap_or_default().to_vec()).into()
    }
}

View on GitHub (pinned to e962003857)

Solutions

  1. Inspect the `.reason` attached to the trc::Error for the exact librdkafka message (often 'Invalid configuration property' or connection details).
  2. Verify the consumer_builder config keys are valid rdkafka properties (e.g. bootstrap.servers, group.id) and that values are correct and reachable.
  3. Ensure librdkafka is installed and the rdkafka crate feature (dynamic vs static) matches your build environment.
  4. Confirm the topic name is a valid Kafka topic string (no whitespace/control characters).

Example fix

// before
let consumer = self.consumer_builder
    .create_with_context(CustomContext) // bootstrap.servers not set
// after
let consumer = self.consumer_builder
    .set("bootstrap.servers", "broker1:9092")
    .set("group.id", "coordinator")
    .create_with_context(CustomContext)
Defensive patterns

Strategy: try-catch

Validate before calling

// check config before subscribe
fn validate_kafka_config(cfg: &HashMap<String, String>) -> Result<(), String> {
    for key in ["bootstrap.servers", "group.id"] {
        if cfg.get(key).map_or(true, |v| v.is_empty()) {
            return Err(format!("missing/empty kafka config key: {}", key));
        }
    }
    Ok(())
}

Try / catch

match pubsub.subscribe(topic).await {
    Ok(stream) => stream,
    Err(e) => {
        tracing::error!(reason = ?e.reason(), "kafka subscriber init failed");
        return Err(e);
    }
}

Prevention

When it happens

Trigger: Calling `NatsKafkaPubSub::subscribe(topic)` when the underlying rdkafka consumer cannot be instantiated: invalid `bootstrap.servers` / client config values, malformed topic- or group-level properties in the consumer_builder, or librdkafka client initialization failure (e.g. missing librdkafka native library, unsupported config key).

Common situations: Wrong or unreachable Kafka bootstrap servers passed into the consumer builder config; a typo'd rdkafka config property (librdkafka rejects unknown keys at creation); running without the librdkafka sys dependency installed; group.id missing or invalid in builder configuration.

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

AI-assisted analysis of stalwartlabs/stalwart@e962003857 (2026-09-06). Data as JSON: /api/errors/6b1bf2c540128408. Report an issue: GitHub.