laminar_connectors/kafka/
latency_profile.rs

1//! Named latency presets over the existing source fetch and producer batching settings.
2//!
3//! Precedence is: connector defaults, then the profile, then explicit options, then validation.
4//! A profile never changes acknowledgements, idempotence, isolation, transactions or the delivery
5//! guarantee, so a faster profile cannot weaken a guarantee.
6
7/// `latency.profile` option shared by the Kafka source and sink.
8#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
9pub enum LatencyProfile {
10    /// Smallest batching delays and per-cycle batches.
11    LowLatency,
12    /// The connector defaults.
13    #[default]
14    Balanced,
15    /// Larger fetches and producer batches with compression.
16    Throughput,
17}
18
19str_enum!(LatencyProfile, lowercase, ConnectorError, "invalid latency.profile",
20    LowLatency => "low_latency", "low-latency";
21    Balanced => "balanced";
22    Throughput => "throughput"
23);
24
25/// Source settings a profile supplies when the option is not set explicitly.
26#[derive(Debug, Clone, Copy, PartialEq, Eq)]
27pub(crate) struct SourceProfileDefaults {
28    pub(crate) max_poll_records: usize,
29    pub(crate) fetch_min_bytes: Option<i32>,
30    pub(crate) fetch_max_wait_ms: Option<i32>,
31}
32
33/// Producer settings a profile supplies when the option is not set explicitly.
34#[derive(Debug, Clone, Copy, PartialEq, Eq)]
35pub(crate) struct SinkProfileDefaults {
36    pub(crate) linger_ms: u64,
37    pub(crate) batch_size: usize,
38    pub(crate) compression: super::CompressionType,
39}
40
41impl LatencyProfile {
42    pub(crate) const fn source_defaults(self) -> SourceProfileDefaults {
43        match self {
44            Self::LowLatency => SourceProfileDefaults {
45                max_poll_records: 256,
46                fetch_min_bytes: Some(1),
47                fetch_max_wait_ms: Some(10),
48            },
49            Self::Balanced => SourceProfileDefaults {
50                max_poll_records: 1000,
51                fetch_min_bytes: None,
52                fetch_max_wait_ms: None,
53            },
54            Self::Throughput => SourceProfileDefaults {
55                max_poll_records: 8192,
56                fetch_min_bytes: Some(64 * 1024),
57                fetch_max_wait_ms: Some(100),
58            },
59        }
60    }
61
62    pub(crate) const fn sink_defaults(self) -> SinkProfileDefaults {
63        match self {
64            Self::LowLatency => SinkProfileDefaults {
65                linger_ms: 0,
66                batch_size: 16_384,
67                compression: super::CompressionType::None,
68            },
69            Self::Balanced => SinkProfileDefaults {
70                linger_ms: 5,
71                batch_size: 16_384,
72                compression: super::CompressionType::None,
73            },
74            Self::Throughput => SinkProfileDefaults {
75                linger_ms: 50,
76                batch_size: 1024 * 1024,
77                compression: super::CompressionType::Lz4,
78            },
79        }
80    }
81}
82
83#[cfg(test)]
84mod tests {
85    use super::*;
86
87    #[test]
88    fn parses_names_and_aliases() {
89        assert_eq!(
90            "low_latency".parse::<LatencyProfile>().unwrap(),
91            LatencyProfile::LowLatency
92        );
93        assert_eq!(
94            "LOW-LATENCY".parse::<LatencyProfile>().unwrap(),
95            LatencyProfile::LowLatency
96        );
97        assert_eq!(
98            "throughput".parse::<LatencyProfile>().unwrap(),
99            LatencyProfile::Throughput
100        );
101        assert!("fastest".parse::<LatencyProfile>().is_err());
102    }
103
104    #[test]
105    fn balanced_is_the_historical_default() {
106        let source = LatencyProfile::Balanced.source_defaults();
107        assert_eq!(source.max_poll_records, 1000);
108        assert_eq!(source.fetch_min_bytes, None);
109        let sink = LatencyProfile::Balanced.sink_defaults();
110        assert_eq!(sink.linger_ms, 5);
111        assert_eq!(sink.batch_size, 16_384);
112    }
113}
laminar_connectors/kafka/latency_profile.rs

laminar_connectors/kafka/
latency_profile.rs
