rainier_framework/
kafka.rs

1//! Building the Kafka pieces from configuration.
2//!
3//! The step between `KAFKA_BROKERS` in an environment file and the objects the
4//! [broadcaster](rainier_broadcast::kafka), the [relay](crate::relay) and the
5//! [queue](rainier_queue::kafka) are built from — so an application writes a
6//! provider rather than assembling a connector by hand.
7//!
8//! ```ignore
9//! // app/providers/kafka.rs
10//! pub fn register(app: &Application) -> Result<()> {
11//!     let config = app.resolve::<Config>()?;
12//!     let client = kafka::client(&config).await?;
13//!
14//!     app.instance(Broadcasting::new(Arc::new(kafka::broadcaster(&config, Arc::clone(&client))?)));
15//!
16//!     relay::spawn(
17//!         kafka::relay(&config, client)?,
18//!         SocketFanOut::new(app.resolve::<Rooms>()?),
19//!     );
20//!     Ok(())
21//! }
22//! ```
23
24use std::sync::Arc;
25
26use rainier_broadcast::kafka::{KafkaBroadcaster, KafkaRelay};
27use rainier_cache::LockManager;
28use rainier_config::Config;
29use rainier_drivers::kafka::{
30    KafkaClient, KafkaConnector, KafkaCredentials, KafkaRecord, SaslMechanism,
31};
32use rainier_queue::KafkaQueue;
33use rainier_support::{Error, Result};
34use serde::Serialize;
35
36use crate::keys;
37
38/// The connector `KAFKA_*` describes, or `None` when no brokers are named.
39///
40/// `None` rather than an error: an application that does not use Kafka should
41/// not have to say so, and the caller decides whether its absence matters.
42pub fn connector(config: &Config) -> Result<Option<KafkaConnector>> {
43    let brokers = config.get(keys::KAFKA_BROKERS).unwrap_or_default();
44
45    if brokers.trim().is_empty() {
46        return Ok(None);
47    }
48
49    let mut connector = KafkaConnector::parse(&brokers)
50        .with_client_id(config.get(keys::APP_NAME).unwrap_or_else(|| "rainier".into()));
51
52    if config.get(keys::KAFKA_TLS).unwrap_or(false) {
53        connector = connector.with_tls();
54    }
55
56    let username = config.get(keys::KAFKA_USERNAME).unwrap_or_default();
57    if !username.is_empty() {
58        let mechanism = mechanism(&config.get(keys::KAFKA_SASL_MECHANISM).unwrap_or_default())?;
59        let password = config.get(keys::KAFKA_PASSWORD).unwrap_or_default();
60
61        connector =
62            connector.with_credentials(KafkaCredentials::new(mechanism, username, password));
63    }
64
65    Ok(Some(connector))
66}
67
68/// Connect to the configured cluster.
69///
70/// # Errors
71///
72/// When `KAFKA_BROKERS` is empty — a provider asking for a client has decided
73/// this application needs one, and starting without it only moves the failure
74/// to the first broadcast.
75pub async fn client(config: &Config) -> Result<Arc<KafkaClient>> {
76    let connector = connector(config)?.ok_or_else(|| {
77        Error::internal(
78            "Kafka is not configured. Set `KAFKA_BROKERS` to a comma-separated list of brokers.",
79        )
80    })?;
81
82    Ok(Arc::new(KafkaClient::connect(&connector).await?))
83}
84
85/// The broadcaster `KAFKA_BROADCAST_TOPIC` and `KAFKA_TOPIC_PREFIX` describe.
86pub fn broadcaster(config: &Config, client: Arc<KafkaClient>) -> KafkaBroadcaster {
87    let topic = topic(config, keys::KAFKA_BROADCAST_TOPIC, "broadcasts");
88
89    KafkaBroadcaster::new(client).on_topic(topic)
90}
91
92/// A relay for the same topic the broadcaster publishes to.
93pub fn relay(config: &Config, client: Arc<KafkaClient>) -> KafkaRelay {
94    KafkaRelay::new(client, topic(config, keys::KAFKA_BROADCAST_TOPIC, "broadcasts"))
95}
96
97/// The queue `KAFKA_GROUP` and `KAFKA_TOPIC_PREFIX` describe.
98///
99/// # Errors
100///
101/// When `locks` is not backed by a shared store — see
102/// [`KafkaQueue::new`](rainier_queue::KafkaQueue::new).
103pub fn queue(config: &Config, client: Arc<KafkaClient>, locks: LockManager) -> Result<KafkaQueue> {
104    let queue = KafkaQueue::new(client, locks)?
105        .in_group(config.get(keys::KAFKA_GROUP).unwrap_or_else(|| "rainier".into()));
106
107    Ok(match config.get(keys::KAFKA_TOPIC_PREFIX) {
108        Some(prefix) if !prefix.is_empty() => queue.with_topic_prefix(prefix),
109        _ => queue,
110    })
111}
112
113/// The record an event becomes on the way to a topic.
114///
115/// The event's own name is a header, so a consumer can route on it without
116/// deserialising, and the body is the event as JSON. Keyed by whatever the
117/// caller says identifies the subject — an account id, an order number — which
118/// is what puts everything about one subject on one partition, in order.
119fn record_for_event<E: Serialize>(name: &str, event: &E, key: String) -> Result<KafkaRecord> {
120    let body = serde_json::to_vec(event)
121        .map_err(|e| Error::internal(format!("an event published to Kafka must serialise: {e}")))?;
122
123    Ok(KafkaRecord::new(body).keyed(key).header("event", name.to_string()))
124}
125
126/// Publish every event of one type to a topic, as it happens.
127///
128/// A broadcast pointed at a log rather than at a browser: the
129/// same event that updates the page becomes a record other services can read.
130///
131/// ```ignore
132/// kafka::publish_events::<OrderShipped>(
133///     &events,
134///     Arc::clone(&client),
135///     "orders",
136///     |event| event.order_id.to_string(),
137/// );
138/// ```
139///
140/// # A failed publish does not fail the request
141///
142/// It is logged and swallowed. A listener that returns `Err` stops the
143/// listeners behind it, and "the analytics topic was unreachable" is not a
144/// reason to abandon the rest of what an event set off. Where the publish
145/// *must* happen, publish from a [job](rainier_queue) instead, so it retries.
146pub fn publish_events<E>(
147    events: &rainier_events::Dispatcher,
148    client: Arc<KafkaClient>,
149    topic: impl Into<String>,
150    key_of: impl Fn(&E) -> String + Send + Sync + 'static,
151) where
152    E: rainier_events::Event + Serialize,
153{
154    let topic = topic.into();
155    let name = E::event_name();
156
157    events.listen::<E, _>(move |event: Arc<E>| {
158        let client = Arc::clone(&client);
159        let topic = topic.clone();
160        let key = key_of(&event);
161
162        Box::pin(async move {
163            match record_for_event(name, &*event, key) {
164                Ok(record) => {
165                    if let Err(e) = client.produce(&topic, vec![record]).await {
166                        tracing::error!(topic, event = name, error = %e, "could not publish an event");
167                    }
168                }
169                Err(e) => tracing::error!(topic, event = name, error = %e, "could not serialise an event"),
170            }
171            Ok(())
172        })
173    });
174}
175
176/// A configured topic name, prefixed.
177fn topic(config: &Config, key: rainier_config::Key<String>, fallback: &str) -> String {
178    let prefix = config.get(keys::KAFKA_TOPIC_PREFIX).unwrap_or_default();
179
180    let name = config
181        .get(key)
182        .filter(|configured| !configured.is_empty())
183        .unwrap_or_else(|| fallback.to_string());
184
185    format!("{prefix}{name}")
186}
187
188/// The mechanism a name refers to.
189///
190/// A closed set, and unknown is an error rather than a default: silently
191/// falling back to `PLAIN` would send a password in the clear because somebody
192/// misspelled `scram-sha-512`.
193fn mechanism(name: &str) -> Result<SaslMechanism> {
194    match name.trim().to_ascii_lowercase().replace('_', "-").as_str() {
195        "" | "plain" => Ok(SaslMechanism::Plain),
196        "scram-sha-256" => Ok(SaslMechanism::ScramSha256),
197        "scram-sha-512" => Ok(SaslMechanism::ScramSha512),
198        other => Err(Error::internal(format!(
199            "`KAFKA_SASL_MECHANISM={other}` is not a mechanism. Use `plain`, `scram-sha-256` or \
200             `scram-sha-512`."
201        ))),
202    }
203}
204
205#[cfg(test)]
206mod tests {
207    use super::*;
208    use rainier_config::Env;
209
210    fn config(env: &str) -> Config {
211        let config = Config::new();
212        let env = Env::parse(env).isolated();
213
214        config.set(keys::KAFKA_BROKERS, env.string("KAFKA_BROKERS", "")).unwrap();
215        config.set(keys::KAFKA_TLS, env.bool("KAFKA_TLS", false)).unwrap();
216        config.set(keys::KAFKA_USERNAME, env.string("KAFKA_USERNAME", "")).unwrap();
217        config.set(keys::KAFKA_PASSWORD, env.string("KAFKA_PASSWORD", "")).unwrap();
218        config
219            .set(keys::KAFKA_SASL_MECHANISM, env.string("KAFKA_SASL_MECHANISM", "plain"))
220            .unwrap();
221        config.set(keys::KAFKA_TOPIC_PREFIX, env.string("KAFKA_TOPIC_PREFIX", "")).unwrap();
222        config
223            .set(keys::KAFKA_BROADCAST_TOPIC, env.string("KAFKA_BROADCAST_TOPIC", "broadcasts"))
224            .unwrap();
225        config.set(keys::APP_NAME, env.string("APP_NAME", "rainier")).unwrap();
226        config
227    }
228
229    #[test]
230    fn no_brokers_is_not_an_error() {
231        // An application that does not use Kafka should not have to say so.
232        assert!(connector(&config("APP_ENV=local")).unwrap().is_none());
233    }
234
235    #[test]
236    fn the_broker_list_becomes_a_connector() {
237        let connector = connector(&config("KAFKA_BROKERS=kafka-1:9092,kafka-2\nAPP_NAME=checkout"))
238            .unwrap()
239            .expect("brokers were configured");
240
241        assert_eq!(connector.brokers(), ["kafka-1:9092", "kafka-2:9092"]);
242        assert!(!connector.is_tls());
243        assert!(connector.credentials().is_none());
244    }
245
246    #[test]
247    fn credentials_are_only_used_when_a_username_is_set() {
248        let with = connector(&config(
249            "KAFKA_BROKERS=kafka:9092\nKAFKA_USERNAME=svc\nKAFKA_PASSWORD=secret\n\
250             KAFKA_SASL_MECHANISM=scram-sha-512",
251        ))
252        .unwrap()
253        .unwrap();
254
255        let credentials = with.credentials().expect("a username was configured");
256        assert_eq!(credentials.mechanism(), SaslMechanism::ScramSha512);
257        assert_eq!(credentials.username(), "svc");
258
259        let without =
260            connector(&config("KAFKA_BROKERS=kafka:9092\nKAFKA_PASSWORD=secret")).unwrap().unwrap();
261        assert!(without.credentials().is_none(), "a password alone authenticates nobody");
262    }
263
264    #[test]
265    fn a_misspelled_mechanism_stops_the_boot() {
266        // The alternative is falling back to PLAIN and sending the password in
267        // the clear because somebody typed `scram-512`.
268        let error = connector(&config(
269            "KAFKA_BROKERS=kafka:9092\nKAFKA_USERNAME=svc\nKAFKA_SASL_MECHANISM=scram-512",
270        ))
271        .unwrap_err();
272
273        assert!(error.message().contains("scram-sha-512"), "{}", error.message());
274    }
275
276    #[test]
277    fn mechanism_names_are_forgiving_about_case_and_underscores() {
278        assert_eq!(mechanism("SCRAM_SHA_256").unwrap(), SaslMechanism::ScramSha256);
279        assert_eq!(mechanism(" plain ").unwrap(), SaslMechanism::Plain);
280        assert_eq!(mechanism("").unwrap(), SaslMechanism::Plain);
281    }
282
283    #[test]
284    fn the_prefix_reaches_the_topic_names() {
285        let config = config("KAFKA_BROKERS=kafka:9092\nKAFKA_TOPIC_PREFIX=checkout.");
286
287        assert_eq!(
288            topic(&config, keys::KAFKA_BROADCAST_TOPIC, "broadcasts"),
289            "checkout.broadcasts"
290        );
291    }
292
293    #[test]
294    fn a_named_broadcast_topic_wins_over_the_default() {
295        let config = config("KAFKA_BROKERS=kafka:9092\nKAFKA_BROADCAST_TOPIC=events");
296
297        assert_eq!(topic(&config, keys::KAFKA_BROADCAST_TOPIC, "broadcasts"), "events");
298    }
299
300    #[test]
301    fn an_event_becomes_a_keyed_record_carrying_its_name() {
302        #[derive(serde::Serialize)]
303        struct OrderShipped {
304            order_id: u64,
305        }
306
307        let record =
308            record_for_event("OrderShipped", &OrderShipped { order_id: 7 }, "7".into()).unwrap();
309
310        assert_eq!(record.key.as_deref(), Some(&b"7"[..]));
311        assert_eq!(record.headers.get("event").map(Vec::as_slice), Some(&b"OrderShipped"[..]));
312
313        // The header exists so a consumer can route without parsing the body,
314        // and the body is still the whole event.
315        let body: serde_json::Value = serde_json::from_slice(&record.value).unwrap();
316        assert_eq!(body["order_id"], 7);
317    }
318
319    #[test]
320    fn tls_is_off_unless_it_is_asked_for() {
321        let plain = connector(&config("KAFKA_BROKERS=kafka:9092")).unwrap().unwrap();
322        assert!(!plain.is_tls());
323
324        let secured =
325            connector(&config("KAFKA_BROKERS=kafka:9092\nKAFKA_TLS=true")).unwrap().unwrap();
326        assert!(secured.is_tls());
327    }
328}
rainier_framework/kafka.rs

rainier_framework/
kafka.rs
