krabka_remote_storage_topic/kafka_log/
config.rs

1//! Construction-time configuration for the Kafka-topic-backed metadata event
2//! log.
3//!
4//! This module holds the topic name, the Apache Kafka defaults for partition
5//! count and replication factor, the transport tunables of the producer and of
6//! the per-partition fetch loops, and the validation that keeps every tunable
7//! inside the `int32` millisecond and byte ranges the Kafka wire carries.
8
9use {
10    krabka_client_core::{
11        ClientFrameMax, ConnectionDispatchQueueCapacity, security::ClientSecurity,
12    },
13    krabka_macros::RefinedNewtype,
14    krabka_units::prelude::{
15        ByteSize, ByteSizeExt as _, Time, TimeExt as _, mebibytes, millis, secs,
16    },
17    refined_type::rule::{GreaterI32, GreaterUsize},
18    std::time::Duration,
19};
20
21/// Default name of the internal metadata topic.
22pub const METADATA_TOPIC: &str = "__remote_log_metadata";
23
24/// Default partition count for `__remote_log_metadata`, matching
25/// Apache Kafka's `remote.log.metadata.topic.num.partitions`.
26pub const DEFAULT_NUM_PARTITIONS: i32 = 50;
27
28/// Default replication factor for `__remote_log_metadata`, matching
29/// Apache Kafka's `remote.log.metadata.topic.replication.factor`.
30pub const DEFAULT_REPLICATION: i32 = 3;
31
32/// Default `min.insync.replicas` for `__remote_log_metadata`, matching
33/// Apache Kafka's `remote.log.metadata.topic.min.isr`
34/// (`TopicBasedRemoteLogMetadataManagerConfig.DEFAULT_REMOTE_LOG_METADATA_TOPIC_MIN_ISR`).
35pub const DEFAULT_MIN_ISR: i32 = 2;
36
37/// How long `CreateTopics` may take to provision `__remote_log_metadata`
38/// before the broker abandons the round-trip.
39pub const DEFAULT_METADATA_TOPIC_CREATE_TIMEOUT: Time = secs(30);
40
41/// `max_wait_ms` for the per-partition metadata `Fetch`. It is long enough
42/// that an idle partition costs one RPC per interval rather than a spin, and
43/// short enough that cancellation on reassignment is prompt.
44pub const DEFAULT_METADATA_FETCH_MAX_WAIT: Time = millis(500);
45
46/// Per-partition budget for the metadata `Fetch`. Metadata events are small,
47/// so one mebibyte is many thousands of them per round-trip.
48pub const DEFAULT_METADATA_FETCH_MAX_BYTES: ByteSize = mebibytes(1);
49
50/// Pause before a retry of a failed metadata `Fetch`, so a broker that is down
51/// does not turn the fetch loop into a busy spin.
52pub const DEFAULT_METADATA_FETCH_RETRY_BACKOFF: Time = millis(200);
53
54/// Default capacity of the shared metadata-event delivery queue.
55pub const DEFAULT_METADATA_EVENT_QUEUE_CAPACITY: usize = 1024;
56
57/// Positive capacity of the shared metadata-event delivery queue.
58#[derive(Clone, Copy, Debug, Eq, PartialEq, RefinedNewtype)]
59#[refined(
60    rule(GreaterUsize<0>),
61    string_error,
62    label = "metadata event queue capacity",
63    getter = capacity,
64    default = DEFAULT_METADATA_EVENT_QUEUE_CAPACITY
65)]
66pub struct MetadataEventQueueCapacity(usize);
67
68/// Construction-time configuration for
69/// [`KafkaMetadataEventLog`](super::KafkaMetadataEventLog).
70#[derive(Debug, Clone)]
71pub struct KafkaMetadataLogConfig {
72    /// `host:port` for the Kafka client to bootstrap from. The TBRLMM in a
73    /// broker connects over loopback to its own listener.
74    pub bootstrap: String,
75    /// Internal topic name. Production deployments keep the default. The
76    /// field exists so multiple isolated clusters can share an environment in
77    /// tests.
78    pub topic: String,
79    /// Number of partitions to create the topic with on first startup.
80    /// The log ignores this value when the topic already exists, and the
81    /// existing count wins. The log does not support re-bucketing on
82    /// partition growth.
83    pub num_partitions: i32,
84    /// Replication factor to create the topic with on first startup.
85    /// The log ignores this value when the topic already exists.
86    pub replication: i32,
87    /// `min.insync.replicas` to create the topic with on first startup.
88    /// `None` leaves the setting to the cluster default. The log ignores this
89    /// value when the topic already exists.
90    pub min_isr: Option<i32>,
91    /// `client_id` for the producer and consumer. It is diagnostic only.
92    pub client_id: String,
93    /// Provision and maintain this internal topic with log compaction.
94    pub compacted: bool,
95    /// Create the topic when it is absent. Read-only tooling disables this.
96    pub provision_topic: bool,
97    /// Client TLS/SASL security applied to the producer, the raw client,
98    /// the admin client, and every per-partition fetch connection.
99    /// `None` is plaintext loopback, and it is the default.
100    pub security: Option<ClientSecurity>,
101    /// Timeout for provisioning the internal topic.
102    pub topic_create_timeout: Time,
103    /// Maximum wait for each per-partition metadata fetch.
104    pub fetch_max_wait: Time,
105    /// Maximum bytes returned by each per-partition metadata fetch.
106    pub fetch_max_bytes: ByteSize,
107    /// Backoff after a failed metadata fetch.
108    pub fetch_retry_backoff: Time,
109    /// Capacity of the shared metadata-event delivery queue.
110    pub event_queue_capacity: MetadataEventQueueCapacity,
111    /// Capacity of every outbound Kafka connection's dispatch queue.
112    pub dispatch_queue_capacity: ConnectionDispatchQueueCapacity,
113    /// Maximum frame size for every outbound Kafka connection.
114    pub frame_max: ClientFrameMax,
115}
116
117impl KafkaMetadataLogConfig {
118    /// Construct a config with the conventional Kafka defaults.
119    #[must_use]
120    pub fn new(bootstrap: impl Into<String>) -> Self {
121        Self {
122            bootstrap: bootstrap.into(),
123            topic: METADATA_TOPIC.to_owned(),
124            num_partitions: DEFAULT_NUM_PARTITIONS,
125            replication: DEFAULT_REPLICATION,
126            min_isr: Some(DEFAULT_MIN_ISR),
127            client_id: "krabka-rlmm".to_owned(),
128            compacted: false,
129            provision_topic: true,
130            security: None,
131            topic_create_timeout: DEFAULT_METADATA_TOPIC_CREATE_TIMEOUT,
132            fetch_max_wait: DEFAULT_METADATA_FETCH_MAX_WAIT,
133            fetch_max_bytes: DEFAULT_METADATA_FETCH_MAX_BYTES,
134            fetch_retry_backoff: DEFAULT_METADATA_FETCH_RETRY_BACKOFF,
135            event_queue_capacity: MetadataEventQueueCapacity::default(),
136            dispatch_queue_capacity: ConnectionDispatchQueueCapacity::default(),
137            frame_max: ClientFrameMax::default(),
138        }
139    }
140
141    /// Validate values that cross into the Kafka wire client.
142    ///
143    /// # Errors
144    ///
145    /// Returns an error for non-positive, non-finite, fractional, or
146    /// out-of-range wire values, and for a `min_isr` below one, as Kafka's
147    /// `remote.log.metadata.topic.min.isr` does (`atLeast(1)`).
148    pub fn validate(&self) -> Result<(), String> {
149        if let Some(min_isr) = self.min_isr {
150            GreaterI32::<0>::new(min_isr).map_err(|error| format!("min_isr: {error}"))?;
151        }
152        validate_positive_whole_millis_i32("topic_create_timeout", self.topic_create_timeout)?;
153        validate_positive_whole_millis_i32("fetch_max_wait", self.fetch_max_wait)?;
154        validate_positive_whole_bytes_i32("fetch_max_bytes", self.fetch_max_bytes)?;
155        validate_positive_duration("fetch_retry_backoff", self.fetch_retry_backoff)
156    }
157}
158
159fn validate_positive_whole_millis_i32(name: &str, value: Time) -> Result<(), String> {
160    let millis = value.millis_i64();
161    if !value.secs_f64().is_finite() || Time::from_millis(millis) != value {
162        return Err(format!(
163            "{name} must be a positive whole number of milliseconds within 1..=i32::MAX"
164        ));
165    }
166    let millis = i32::try_from(millis).map_err(|_| {
167        format!("{name} must be a positive whole number of milliseconds within 1..=i32::MAX")
168    })?;
169    GreaterI32::<0>::new(millis)
170        .map(|_| ())
171        .map_err(|error| format!("{name}: {error}"))
172}
173
174fn validate_positive_whole_bytes_i32(name: &str, value: ByteSize) -> Result<(), String> {
175    let bytes = value.bytes_i64();
176    if !value.bytes_f64().is_finite() || ByteSize::from_bytes_i64(bytes) != value {
177        return Err(format!(
178            "{name} must be a positive whole number of bytes within 1..=i32::MAX"
179        ));
180    }
181    let bytes = i32::try_from(bytes).map_err(|_| {
182        format!("{name} must be a positive whole number of bytes within 1..=i32::MAX")
183    })?;
184    GreaterI32::<0>::new(bytes)
185        .map(|_| ())
186        .map_err(|error| format!("{name}: {error}"))
187}
188
189fn validate_positive_duration(name: &str, value: Time) -> Result<(), String> {
190    let duration = Duration::try_from_secs_f64(value.secs_f64())
191        .map_err(|error| format!("{name}: {error}"))?;
192    if duration.is_zero() {
193        return Err(format!("{name} must be positive"));
194    }
195    Ok(())
196}
197
198#[cfg(test)]
199mod tests {
200    use {
201        super::*,
202        assert2::{assert, check},
203        std::time::Duration,
204    };
205
206    fn configured(configure: impl FnOnce(&mut KafkaMetadataLogConfig)) -> KafkaMetadataLogConfig {
207        let mut cfg = KafkaMetadataLogConfig::new("127.0.0.1:9092");
208        configure(&mut cfg);
209        cfg
210    }
211
212    #[test]
213    fn config_defaults_match_kafka() {
214        let cfg = KafkaMetadataLogConfig::new("127.0.0.1:9092");
215        check!(cfg.topic == METADATA_TOPIC);
216        check!(cfg.num_partitions == 50);
217        check!(cfg.replication == 3);
218        check!(cfg.min_isr == Some(2));
219        check!(!cfg.compacted);
220        check!(cfg.provision_topic);
221        check!(cfg.bootstrap == "127.0.0.1:9092");
222        check!(cfg.security.is_none());
223        check!(cfg.topic_create_timeout == secs(30));
224        check!(cfg.fetch_max_wait == millis(500));
225        check!(cfg.fetch_max_bytes == mebibytes(1));
226        check!(cfg.fetch_retry_backoff == millis(200));
227        check!(cfg.event_queue_capacity.capacity() == 1024);
228        cfg.validate().unwrap();
229    }
230
231    #[test]
232    fn config_accepts_custom_transport_policy() {
233        let cfg = KafkaMetadataLogConfig {
234            topic_create_timeout: secs(45),
235            fetch_max_wait: millis(750),
236            fetch_max_bytes: mebibytes(2),
237            fetch_retry_backoff: millis(300),
238            event_queue_capacity: MetadataEventQueueCapacity::new(2048).unwrap(),
239            ..KafkaMetadataLogConfig::new("127.0.0.1:9092")
240        };
241
242        cfg.validate().unwrap();
243        check!(cfg.topic_create_timeout == secs(45));
244        check!(cfg.fetch_max_wait == millis(750));
245        check!(cfg.fetch_max_bytes == mebibytes(2));
246        check!(cfg.fetch_retry_backoff == millis(300));
247        check!(cfg.event_queue_capacity.capacity() == 2048);
248    }
249
250    #[yare::parameterized(
251        create_timeout_zero = {
252            "topic_create_timeout",
253            configured(|cfg| cfg.topic_create_timeout = Time::ZERO),
254        },
255        create_timeout_submillisecond = {
256            "topic_create_timeout",
257            configured(|cfg| cfg.topic_create_timeout = Time::from_micros(500)),
258        },
259        create_timeout_infinite = {
260            "topic_create_timeout",
261            configured(|cfg| cfg.topic_create_timeout = Time::from_secs_f64(f64::INFINITY)),
262        },
263        create_timeout_too_large = {
264            "topic_create_timeout",
265            configured(|cfg| cfg.topic_create_timeout = Time::from_millis(i64::from(i32::MAX) + 1)),
266        },
267        fetch_wait_zero = { "fetch_max_wait", configured(|cfg| cfg.fetch_max_wait = Time::ZERO) },
268        fetch_wait_submillisecond = {
269            "fetch_max_wait",
270            configured(|cfg| cfg.fetch_max_wait = Time::from_micros(500)),
271        },
272        fetch_wait_infinite = {
273            "fetch_max_wait",
274            configured(|cfg| cfg.fetch_max_wait = Time::from_secs_f64(f64::INFINITY)),
275        },
276        fetch_wait_too_large = {
277            "fetch_max_wait",
278            configured(|cfg| cfg.fetch_max_wait = Time::from_millis(i64::from(i32::MAX) + 1)),
279        },
280        fetch_bytes_zero = {
281            "fetch_max_bytes",
282            configured(|cfg| cfg.fetch_max_bytes = ByteSize::ZERO),
283        },
284        fetch_bytes_fractional = {
285            "fetch_max_bytes",
286            configured(|cfg| cfg.fetch_max_bytes = ByteSize::from_bytes_f64(0.5)),
287        },
288        fetch_bytes_infinite = {
289            "fetch_max_bytes",
290            configured(|cfg| cfg.fetch_max_bytes = ByteSize::from_bytes_f64(f64::INFINITY)),
291        },
292        fetch_bytes_too_large = {
293            "fetch_max_bytes",
294            configured(|cfg| cfg.fetch_max_bytes = ByteSize::from_bytes_i64(i64::from(i32::MAX) + 1)),
295        },
296        retry_backoff_zero = {
297            "fetch_retry_backoff",
298            configured(|cfg| cfg.fetch_retry_backoff = Time::ZERO),
299        },
300        retry_backoff_infinite = {
301            "fetch_retry_backoff",
302            configured(|cfg| cfg.fetch_retry_backoff = Time::from_secs_f64(f64::INFINITY)),
303        },
304    )]
305    fn config_rejects_invalid_transport_policy(field: &str, cfg: KafkaMetadataLogConfig) {
306        let error = cfg.validate().expect_err("invalid policy must fail");
307        assert2::assert!(error.contains(field), "field={field}, error={error}");
308    }
309
310    #[yare::parameterized(
311        one = { Some(1), true },
312        two = { Some(2), true },
313        zero = { Some(0), false },
314        negative = { Some(-1), false },
315        unset_uses_cluster_default = { None, true },
316    )]
317    fn min_isr_must_be_at_least_one(min_isr: Option<i32>, valid: bool) {
318        let cfg = KafkaMetadataLogConfig {
319            min_isr,
320            ..KafkaMetadataLogConfig::new("127.0.0.1:9092")
321        };
322        let result = cfg.validate();
323        assert2::assert!(result.is_ok() == valid, "min_isr={min_isr:?}: {result:?}");
324        if let Err(error) = result {
325            assert2::assert!(error.contains("min_isr"), "{error}");
326        }
327    }
328
329    #[test]
330    fn metadata_event_queue_capacity_rejects_zero() {
331        assert!(MetadataEventQueueCapacity::new(0).is_err());
332        check!(MetadataEventQueueCapacity::new(1).unwrap().capacity() == 1);
333    }
334
335    /// The metadata client's tunables are quantities, but they reach Kafka as
336    /// raw `int32` milliseconds and bytes. This test pins the wire images. A
337    /// wrong scale here is invisible in the types. It would show up only as a
338    /// metadata consumer that spins, when `max_wait` is too short, or that
339    /// truncates batches.
340    #[test]
341    fn client_tunables_convert_to_their_kafka_wire_images() {
342        let cfg = KafkaMetadataLogConfig::new("127.0.0.1:9092");
343        check!(cfg.topic_create_timeout.millis_i32() == 30_000);
344        check!(cfg.fetch_max_wait.millis_i32() == 500);
345        check!(cfg.fetch_max_bytes.bytes_i32() == 1 << 20);
346        check!(cfg.fetch_retry_backoff.to_std() == Duration::from_millis(200));
347    }
348
349    #[test]
350    fn config_carries_client_resource_policy() {
351        let cfg = KafkaMetadataLogConfig {
352            dispatch_queue_capacity: ConnectionDispatchQueueCapacity::new(7).unwrap(),
353            frame_max: ClientFrameMax::try_from(krabka_units::kibibytes(32)).unwrap(),
354            ..KafkaMetadataLogConfig::new("127.0.0.1:9092")
355        };
356        check!(cfg.dispatch_queue_capacity.get() == 7);
357        check!(cfg.frame_max.size() == krabka_units::kibibytes(32));
358    }
359
360    #[test]
361    fn config_carries_security() {
362        use {
363            krabka_client_core::security::{ClientSecurity, SaslCredentials},
364            krabka_security::ListenerProtocol,
365        };
366        let cfg = KafkaMetadataLogConfig {
367            bootstrap: "127.0.0.1:9092".into(),
368            topic: METADATA_TOPIC.into(),
369            num_partitions: 1,
370            replication: 1,
371            client_id: "x".into(),
372            security: Some(ClientSecurity {
373                protocol: ListenerProtocol::SaslPlaintext,
374                tls: None,
375                sasl: Some(SaslCredentials::Plain {
376                    username: "u".into(),
377                    password: "p".into(),
378                }),
379                sasl_host: None,
380            }),
381            ..KafkaMetadataLogConfig::new("127.0.0.1:9092")
382        };
383        assert!(cfg.security.is_some());
384    }
385}
krabka_remote_storage_topic/kafka_log/config.rs

krabka_remote_storage_topic/kafka_log/
config.rs
