What causes "Cluster::PublisherError" in stalwartlabs/stalwart?

Inspect err from the mapped PublisherError for the rdkafka root cause.

Cluster::PublisherError

Error message

Cluster::PublisherError

What it means

The Kafka-backed PubSub publisher fails to enqueue a message: rdkafka's producer returned a (error, owned-record) tuple, which is mapped into a trc ClusterEvent::PublisherError. The message was not published to the topic.

Source

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

};
use std::time::Duration;
use trc::{ClusterEvent, Error, EventType};

pub struct KafkaPubSubStream {
    subs: LoggingConsumer,
}

impl KafkaPubSub {
    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 }))
    }
}

View on GitHub (pinned to e962003857)

Solutions

Inspect err from the mapped PublisherError for the rdkafka root cause.
Verify bootstrap brokers, SASL/TLS settings and topic existence/permissions.
Retry publishing; consider a non-zero timeout instead of Duration::from_secs(0) to tolerate a briefly full queue.
Monitor producer queue usage and increase queue size/linger if messages are dropped under load.
Check broker health/consumer group state and network reachability from the host.

Example fix

// before
producer.send(FutureRecord::to(topic).payload(msg), Duration::from_secs(0)).await
// after
producer.send(FutureRecord::to(topic).payload(msg), Duration::from_secs(5)).await

Defensive patterns

Strategy: retry

Validate before calling

// before publishing, verify broker connectivity
// rdkafka health check or: kafka-topics --bootstrap-server $BROKERS --describe --topic $TOPIC

Try / catch

for attempt in 0..3 {
    match publisher.publish(topic, msg).await {
        Ok(()) => break,
        Err(err) if attempt < 2 => tokio::time::sleep(backoff(attempt)).await,
        Err(err) => return Err(err),
    }
}

Prevention

Monitor Kafka broker availability and producer error metrics.
Use a non-zero delivery timeout so a full queue waits instead of rejecting.
Size the producer queue for peak publish rates.
Validate topic existence and ACLs during deployment.
Test SASL/TLS credentials before production rollout.

When it happens

Trigger: Calling `publish` when the Kafka producer's send fails — broker unreachable, message queue full (timeout 0s means immediate rejection when local queue is full), authentication failure, or invalid topic.

Common situations: Kafka broker down or wrong bootstrap servers; SASL credentials wrong; local producer queue saturated under load (delivery timeout of 0 gives no wait); topic deleted or authorization denied.

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

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