{"record":{"id":"a13df4a9b4901981","repo":"stalwartlabs/stalwart","slug":"cluster-publishererror","errorCode":null,"errorMessage":"Cluster::PublisherError","messagePattern":"Cluster::PublisherError","errorType":"error_code","errorClass":null,"httpStatus":null,"severity":"error","filePath":"crates/coordinator/src/backend/kafka/pubsub.rs","lineNumber":31,"sourceCode":"};\nuse std::time::Duration;\nuse trc::{ClusterEvent, Error, EventType};\n\npub struct KafkaPubSubStream {\n    subs: LoggingConsumer,\n}\n\nimpl KafkaPubSub {\n    pub async fn publish(&self, topic: &'static str, message: Vec<u8>) -> trc::Result<()> {\n        self.producer\n            .send(\n                FutureRecord::<(), [u8]>::to(topic).payload(message.as_slice()),\n                Duration::from_secs(0),\n            )\n            .await\n            .map(|_| ())\n            .map_err(|(err, _)| {\n                Error::new(EventType::Cluster(ClusterEvent::PublisherError)).reason(err)\n            })\n    }\n\n    pub async fn subscribe(&self, topic: &'static str) -> trc::Result<PubSubStream> {\n        let subs: StreamConsumer<CustomContext> = self\n            .consumer_builder\n            .create_with_context(CustomContext)\n            .map_err(|err| {\n                Error::new(EventType::Cluster(ClusterEvent::SubscriberError)).reason(err)\n            })?;\n        subs.subscribe(&[topic]).map_err(|err| {\n            Error::new(EventType::Cluster(ClusterEvent::SubscriberError)).reason(err)\n        })?;\n\n        Ok(PubSubStream::Kafka(KafkaPubSubStream { subs }))\n    }\n}\n","sourceCodeStart":13,"sourceCodeEnd":49,"githubUrl":"https://github.com/stalwartlabs/stalwart/blob/e96200385781a6a9995a8b839ac27d6c75a983ee/crates/coordinator/src/backend/kafka/pubsub.rs#L13-L49","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","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."],"exampleFix":"// before\nproducer.send(FutureRecord::to(topic).payload(msg), Duration::from_secs(0)).await\n// after\nproducer.send(FutureRecord::to(topic).payload(msg), Duration::from_secs(5)).await","handlingStrategy":"retry","validationCode":"// before publishing, verify broker connectivity\n// rdkafka health check or: kafka-topics --bootstrap-server $BROKERS --describe --topic $TOPIC","typeGuard":null,"tryCatchPattern":"for attempt in 0..3 {\n    match publisher.publish(topic, msg).await {\n        Ok(()) => break,\n        Err(err) if attempt < 2 => tokio::time::sleep(backoff(attempt)).await,\n        Err(err) => return Err(err),\n    }\n}","preventionTips":["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."],"tags":["kafka","pubsub","network","messaging"],"backgroundTag":"http-request-failed","analyzedSha":"e96200385781a6a9995a8b839ac27d6c75a983ee","analyzedAt":"2026-09-06T22:07:17.982Z","contentChangedAt":"2026-09-06T22:07:17.982Z","schemaVersion":2},"datasetVersion":"2026-09-14T05:17:10.506Z"}