laminar_connectors/kafka/config/
source_options.rs

1//! Parsing of Kafka source connector properties.
2//!
3//! Each phase fills one coherent part of `KafkaSourceConfig`. Validation runs only after every
4//! phase succeeds, so callers never observe a partially prepared configuration.
5
6use std::time::Duration;
7
8use crate::config::ConnectorConfig;
9use crate::error::ConnectorError;
10use crate::serde::Format;
11
12use super::{
13    parse_specific_offsets, AssignmentStrategy, IsolationLevel, KafkaSourceConfig, OffsetReset,
14    SchemaEvolutionStrategy, SecurityProtocol, SrAuth, StartupMode, SubjectNameStrategy,
15    TopicSubscription,
16};
17
18impl KafkaSourceConfig {
19    /// Parses a Kafka source configuration from connector properties.
20    ///
21    /// # Errors
22    ///
23    /// Returns an error when a required property is missing or any value is invalid.
24    #[allow(deprecated)]
25    pub fn from_config(config: &ConnectorConfig) -> Result<Self, ConnectorError> {
26        let mut parsed = Self::default();
27        parsed.parse_connection(config)?;
28        parsed.parse_schema(config)?;
29        parsed.parse_startup_and_fetch(config)?;
30        parsed.parse_group_timing(config)?;
31        parsed.parse_delivery_controls(config)?;
32        parsed.kafka_properties = config.properties_with_prefix("kafka.");
33        parsed.validate()?;
34        Ok(parsed)
35    }
36
37    fn parse_connection(&mut self, config: &ConnectorConfig) -> Result<(), ConnectorError> {
38        self.bootstrap_servers = config.require("bootstrap.servers")?.to_string();
39        self.group_id = config.require("group.id")?.to_string();
40        self.subscription = if let Some(pattern) = config.get("topic.pattern") {
41            TopicSubscription::Pattern(pattern.to_string())
42        } else {
43            let topics = config
44                .require("topic")?
45                .split(',')
46                .map(|topic| topic.trim().to_string())
47                .collect();
48            TopicSubscription::Topics(topics)
49        };
50
51        self.security_protocol = config
52            .get("security.protocol")
53            .map_or(Ok(SecurityProtocol::default()), str::parse)?;
54        self.sasl_mechanism = config.get("sasl.mechanism").map(str::parse).transpose()?;
55        self.sasl_username = config.get("sasl.username").map(String::from);
56        self.sasl_password = config.get("sasl.password").map(String::from);
57        self.ssl_ca_location = config.get("ssl.ca.location").map(String::from);
58        self.ssl_certificate_location = config.get("ssl.certificate.location").map(String::from);
59        self.ssl_key_location = config.get("ssl.key.location").map(String::from);
60        self.ssl_key_password = config.get("ssl.key.password").map(String::from);
61        Ok(())
62    }
63
64    fn parse_schema(&mut self, config: &ConnectorConfig) -> Result<(), ConnectorError> {
65        self.format = config.get("format").map_or(Ok(Format::Json), |format| {
66            format
67                .parse::<Format>()
68                .map_err(|error| ConnectorError::ConfigurationError(error.to_string()))
69        })?;
70        self.schema_registry_url = config.get("schema.registry.url").map(String::from);
71        self.schema_registry_auth = parse_schema_registry_auth(config)?;
72        self.schema_compatibility = config
73            .get("schema.compatibility")
74            .map(str::parse)
75            .transpose()?;
76        self.schema_evolution_strategy = config
77            .get("schema.evolution.strategy")
78            .map_or(Ok(SchemaEvolutionStrategy::default()), str::parse)?;
79        self.schema_registry_ssl_ca_location = config
80            .get("schema.registry.ssl.ca.location")
81            .map(String::from);
82        self.schema_registry_ssl_certificate_location = config
83            .get("schema.registry.ssl.certificate.location")
84            .map(String::from);
85        self.schema_registry_ssl_key_location = config
86            .get("schema.registry.ssl.key.location")
87            .map(String::from);
88        self.schema_registry_subject_strategy = config
89            .get("schema.registry.subject.name.strategy")
90            .map_or(Ok(SubjectNameStrategy::default()), str::parse)?;
91        self.schema_registry_record_name =
92            config.get("schema.registry.record.name").map(String::from);
93        validate_subject_name_strategy(
94            self.schema_registry_subject_strategy,
95            self.schema_registry_record_name.as_deref(),
96        )?;
97        self.schema_registry_discovery_timeout = config
98            .get_parsed::<u64>("schema.registry.discovery.timeout.ms")?
99            .map_or(Duration::from_secs(10), Duration::from_millis);
100        self.include_metadata = config
101            .get_parsed::<bool>("include.metadata")?
102            .unwrap_or(false);
103        self.include_headers = config
104            .get_parsed::<bool>("include.headers")?
105            .unwrap_or(false);
106        Ok(())
107    }
108
109    fn parse_startup_and_fetch(&mut self, config: &ConnectorConfig) -> Result<(), ConnectorError> {
110        self.startup_mode = parse_startup_mode(config)?;
111        self.auto_offset_reset = match &self.startup_mode {
112            StartupMode::Earliest => OffsetReset::Earliest,
113            StartupMode::Latest => OffsetReset::Latest,
114            _ => config
115                .get("auto.offset.reset")
116                .map_or(Ok(OffsetReset::Earliest), str::parse)?,
117        };
118        self.isolation_level = config
119            .get("isolation.level")
120            .map_or(Ok(IsolationLevel::default()), str::parse)?;
121        self.max_poll_records = config
122            .get_parsed::<usize>("max.poll.records")?
123            .unwrap_or(1000);
124        self.partition_assignment_strategy = config
125            .get("partition.assignment.strategy")
126            .map_or(Ok(AssignmentStrategy::Range), str::parse)?;
127        self.fetch_min_bytes = config.get_parsed::<i32>("fetch.min.bytes")?;
128        self.fetch_max_bytes = config.get_parsed::<i32>("fetch.max.bytes")?;
129        self.fetch_max_wait_ms = config.get_parsed::<i32>("fetch.max.wait.ms")?;
130        self.max_partition_fetch_bytes = config.get_parsed::<i32>("max.partition.fetch.bytes")?;
131        Ok(())
132    }
133
134    fn parse_group_timing(&mut self, config: &ConnectorConfig) -> Result<(), ConnectorError> {
135        let session_timeout_ms = config
136            .get_parsed::<u64>("session.timeout.ms")?
137            .unwrap_or(45_000);
138        let heartbeat_interval_ms = config
139            .get_parsed::<u64>("heartbeat.interval.ms")?
140            .unwrap_or(10_000);
141        self.queued_max_messages_kbytes = config
142            .get_parsed::<u32>("queued.max.messages.kbytes")?
143            .unwrap_or(16384);
144        let max_poll_interval_ms = config
145            .get_parsed::<u64>("max.poll.interval.ms")?
146            .unwrap_or(600_000);
147
148        self.session_timeout = Duration::from_millis(session_timeout_ms);
149        self.heartbeat_interval = Duration::from_millis(heartbeat_interval_ms);
150        self.max_poll_interval = Duration::from_millis(max_poll_interval_ms);
151        Ok(())
152    }
153
154    fn parse_delivery_controls(&mut self, config: &ConnectorConfig) -> Result<(), ConnectorError> {
155        reject_deprecated_commit_interval(config)?;
156        self.broker_commit_on_checkpoint = config
157            .get_parsed::<bool>("broker.commit.on.checkpoint")?
158            .unwrap_or(true);
159        self.reader_channel_capacity = config
160            .get_parsed::<usize>("reader.channel.capacity")?
161            .unwrap_or(8192);
162        self.backpressure_high_watermark = config
163            .get_parsed::<f64>("backpressure.high.watermark")?
164            .unwrap_or(0.8);
165        self.backpressure_low_watermark = config
166            .get_parsed::<f64>("backpressure.low.watermark")?
167            .unwrap_or(0.25);
168        self.max_deser_error_rate = config
169            .get_parsed::<f64>("max.deser.error.rate")?
170            .unwrap_or(0.5);
171        Ok(())
172    }
173}
174
175fn parse_schema_registry_auth(config: &ConnectorConfig) -> Result<Option<SrAuth>, ConnectorError> {
176    match (
177        config.get("schema.registry.username"),
178        config.get("schema.registry.password"),
179    ) {
180        (Some(username), Some(password)) => Ok(Some(SrAuth {
181            username: username.to_string(),
182            password: password.to_string(),
183        })),
184        (Some(_), None) | (None, Some(_)) => Err(ConnectorError::ConfigurationError(
185            "schema.registry.username and schema.registry.password must both be set".to_string(),
186        )),
187        (None, None) => Ok(None),
188    }
189}
190
191fn validate_subject_name_strategy(
192    strategy: SubjectNameStrategy,
193    record_name: Option<&str>,
194) -> Result<(), ConnectorError> {
195    if matches!(
196        strategy,
197        SubjectNameStrategy::RecordName | SubjectNameStrategy::TopicRecordName
198    ) && record_name.is_none()
199    {
200        return Err(ConnectorError::ConfigurationError(format!(
201            "schema.registry.subject.name.strategy={strategy} requires schema.registry.record.name"
202        )));
203    }
204    Ok(())
205}
206
207fn parse_startup_mode(config: &ConnectorConfig) -> Result<StartupMode, ConnectorError> {
208    if let Some(offsets) = config.get("startup.specific.offsets") {
209        return Ok(StartupMode::SpecificOffsets(parse_specific_offsets(
210            offsets,
211        )?));
212    }
213    if let Some(timestamp) = config.get("startup.timestamp.ms") {
214        let timestamp = timestamp.parse().map_err(|_| {
215            ConnectorError::ConfigurationError(format!(
216                "invalid startup.timestamp.ms: '{timestamp}'"
217            ))
218        })?;
219        return Ok(StartupMode::Timestamp(timestamp));
220    }
221    config
222        .get("startup.mode")
223        .map_or(Ok(StartupMode::default()), str::parse)
224}
225
226fn reject_deprecated_commit_interval(config: &ConnectorConfig) -> Result<(), ConnectorError> {
227    if config
228        .properties()
229        .contains_key("broker.commit.interval.ms")
230    {
231        return Err(ConnectorError::ConfigurationError(
232            "broker.commit.interval.ms is no longer supported — broker offset commits now happen \
233             on checkpoint completion. Set broker.commit.on.checkpoint=false to disable."
234                .into(),
235        ));
236    }
237    Ok(())
238}
laminar_connectors/kafka/config/source_options.rs

laminar_connectors/kafka/config/
source_options.rs
