rightsize_modules/
kafka.rs

1//! A single-node Kafka broker (KRaft mode, no ZooKeeper).
2
3use rightsize::{Container, ContainerGuard, Result, Wait};
4
5/// A single-node Kafka broker.
6pub struct KafkaContainer(Container);
7
8impl KafkaContainer {
9    const PORT: u16 = 9092;
10
11    /// Builds a container from the pinned default image (`apache/kafka:4.0.0`).
12    pub fn new() -> Self {
13        Self::with_image("apache/kafka:4.0.0")
14    }
15
16    /// Builds a container from a caller-chosen image.
17    pub fn with_image(image: &str) -> Self {
18        let container = Container::new(image)
19            .with_exposed_ports(&[Self::PORT])
20            .with_env("KAFKA_NODE_ID", "1")
21            .with_env("KAFKA_PROCESS_ROLES", "broker,controller")
22            .with_env("KAFKA_CONTROLLER_QUORUM_VOTERS", "1@localhost:9091")
23            .with_env(
24                "KAFKA_LISTENERS",
25                "PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9091",
26            )
27            .with_env("KAFKA_CONTROLLER_LISTENER_NAMES", "CONTROLLER")
28            .with_env(
29                "KAFKA_LISTENER_SECURITY_PROTOCOL_MAP",
30                "PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT",
31            )
32            .with_env("KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR", "1")
33            .with_env("KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS", "0")
34            // The apache/kafka image defaults KAFKA_HEAP_OPTS to -Xmx1G, which
35            // exceeds microsandbox's default microVM RAM (~450M) and aborts the
36            // JVM ("insufficient memory"). A single-node KRaft dev broker runs
37            // comfortably in a 256M heap; harmless on the Docker backend, which isn't
38            // memory-constrained here.
39            .with_env("KAFKA_HEAP_OPTS", "-Xmx256M -Xms256M")
40            .waiting_for(Wait::for_log_message(".*Kafka Server started.*", 1))
41            // Rewrites the advertised listener to carry the mapped host port; see
42            // RedpandaContainer's customizer for why this needs the `mapped`
43            // callback.
44            .with_spec_customizer(|mut spec, mapped| {
45                spec.env.push((
46                    "KAFKA_ADVERTISED_LISTENERS".to_string(),
47                    format!("PLAINTEXT://127.0.0.1:{}", mapped(Self::PORT)),
48                ));
49                spec
50            });
51        Self(container)
52    }
53
54    /// Boots the container.
55    pub async fn start(self) -> Result<KafkaGuard> {
56        crate::register_default_backends();
57        Ok(KafkaGuard(self.0.start().await?))
58    }
59}
60
61impl Default for KafkaContainer {
62    fn default() -> Self {
63        Self::new()
64    }
65}
66
67/// The running guard for a [`KafkaContainer`].
68pub struct KafkaGuard(ContainerGuard);
69
70impl KafkaGuard {
71    /// The `PLAINTEXT://` bootstrap-servers address for the running broker.
72    pub fn bootstrap_servers(&self) -> String {
73        format!(
74            "PLAINTEXT://{}:{}",
75            self.0.host(),
76            self.0.get_mapped_port(KafkaContainer::PORT).unwrap()
77        )
78    }
79
80    /// Stops and removes the container, releasing its host port.
81    pub async fn stop(self) -> Result<()> {
82        self.0.stop().await
83    }
84}
85
86impl std::ops::Deref for KafkaGuard {
87    type Target = ContainerGuard;
88    fn deref(&self) -> &ContainerGuard {
89        &self.0
90    }
91}
92
93#[cfg(test)]
94mod tests {
95    use super::*;
96
97    #[test]
98    fn with_image_smoke() {
99        let _ = KafkaContainer::new();
100    }
101}
rightsize_modules/kafka.rs

rightsize_modules/
kafka.rs
