{"record":{"id":"6b1bf2c540128408","repo":"stalwartlabs/stalwart","slug":"cluster-subscribererror","errorCode":null,"errorMessage":"Cluster::SubscriberError","messagePattern":"Cluster::SubscriberError","errorType":"error_code","errorClass":null,"httpStatus":null,"severity":"error","filePath":"crates/coordinator/src/backend/kafka/pubsub.rs","lineNumber":40,"sourceCode":"    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\nimpl KafkaPubSubStream {\n    pub async fn next(&mut self) -> Option<Msg> {\n        let msg = self.subs.recv().await.ok()?;\n        let _ = self.subs.commit_message(&msg, CommitMode::Async);\n        Msg::Kafka(msg.payload().unwrap_or_default().to_vec()).into()\n    }\n}\n","sourceCodeStart":22,"sourceCodeEnd":57,"githubUrl":"https://github.com/stalwartlabs/stalwart/blob/e96200385781a6a9995a8b839ac27d6c75a983ee/crates/coordinator/src/backend/kafka/pubsub.rs#L22-L57","documentation":"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)`.","triggerScenarios":"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).","commonSituations":"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.","solutions":["Inspect the `.reason` attached to the trc::Error for the exact librdkafka message (often 'Invalid configuration property' or connection details).","Verify the consumer_builder config keys are valid rdkafka properties (e.g. bootstrap.servers, group.id) and that values are correct and reachable.","Ensure librdkafka is installed and the rdkafka crate feature (dynamic vs static) matches your build environment.","Confirm the topic name is a valid Kafka topic string (no whitespace/control characters)."],"exampleFix":"// before\nlet consumer = self.consumer_builder\n    .create_with_context(CustomContext) // bootstrap.servers not set\n// after\nlet consumer = self.consumer_builder\n    .set(\"bootstrap.servers\", \"broker1:9092\")\n    .set(\"group.id\", \"coordinator\")\n    .create_with_context(CustomContext)","handlingStrategy":"try-catch","validationCode":"// check config before subscribe\nfn validate_kafka_config(cfg: &HashMap<String, String>) -> Result<(), String> {\n    for key in [\"bootstrap.servers\", \"group.id\"] {\n        if cfg.get(key).map_or(true, |v| v.is_empty()) {\n            return Err(format!(\"missing/empty kafka config key: {}\", key));\n        }\n    }\n    Ok(())\n}","typeGuard":null,"tryCatchPattern":"match pubsub.subscribe(topic).await {\n    Ok(stream) => stream,\n    Err(e) => {\n        tracing::error!(reason = ?e.reason(), \"kafka subscriber init failed\");\n        return Err(e);\n    }\n}","preventionTips":["Validate bootstrap.servers and group.id at startup with a smoke-test consumer.","Pin and test librdkafka availability in CI for your target platforms.","Log the full rdkafka reason string, not just the Cluster::SubscriberError wrapper."],"tags":["kafka","pubsub","subscriber","configuration"],"backgroundTag":"connection-refused","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"}