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, GroupProtocol, IsolationLevel,
14    KafkaMetadataColumns, KafkaSourceConfig, LatencyProfile, OffsetReset, SchemaEvolutionStrategy,
15    SecurityProtocol, SrAuth, StartupMode, SubjectNameStrategy, TopicSubscription,
16};
17
18/// Classic-protocol settings that KIP-848 moves to the broker.
19const CLASSIC_GROUP_OPTIONS: [&str; 3] = [
20    "session.timeout.ms",
21    "heartbeat.interval.ms",
22    "partition.assignment.strategy",
23];
24
25impl KafkaSourceConfig {
26    /// Parses a Kafka source configuration from connector properties.
27    ///
28    /// # Errors
29    ///
30    /// Returns an error when a required property is missing or any value is invalid.
31    #[allow(deprecated)]
32    pub fn from_config(config: &ConnectorConfig) -> Result<Self, ConnectorError> {
33        if config.get("replay.order").is_some() {
34            return Err(ConnectorError::ConfigurationError(
35                "Kafka replay.order is unsupported; cross-partition replay ordering belongs to the engine".into(),
36            ));
37        }
38        if config.get("envelope").is_some() {
39            return Err(ConnectorError::ConfigurationError(
40                "Kafka sources have no 'envelope' option: a table source keyed by the message key \
41                 (key + value = upsert, key + NULL = delete) is not supported; use \
42                 format = 'debezium' for keyed changes"
43                    .into(),
44            ));
45        }
46        let mut parsed = Self::default();
47        parsed.parse_connection(config)?;
48        parsed.parse_schema(config)?;
49        parsed.parse_startup_and_fetch(config)?;
50        parsed.parse_group_timing(config)?;
51        parsed.parse_delivery_controls(config)?;
52        parsed.kafka_properties = config.properties_with_prefix("kafka.");
53        parsed.validate()?;
54        Ok(parsed)
55    }
56
57    fn parse_connection(&mut self, config: &ConnectorConfig) -> Result<(), ConnectorError> {
58        self.bootstrap_servers = config.require("bootstrap.servers")?.to_string();
59        self.group_id = config.require("group.id")?.to_string();
60        self.subscription = if let Some(pattern) = config.get("topic.pattern") {
61            TopicSubscription::Pattern(pattern.to_string())
62        } else {
63            let topics = config
64                .require("topic")?
65                .split(',')
66                .map(|topic| topic.trim().to_string())
67                .collect();
68            TopicSubscription::Topics(topics)
69        };
70
71        self.security_protocol = config
72            .get("security.protocol")
73            .map_or(Ok(SecurityProtocol::default()), str::parse)?;
74        self.sasl_mechanism = config.get("sasl.mechanism").map(str::parse).transpose()?;
75        self.sasl_username = config.get("sasl.username").map(String::from);
76        self.sasl_password = config.get("sasl.password").map(String::from);
77        self.ssl_ca_location = config.get("ssl.ca.location").map(String::from);
78        self.ssl_certificate_location = config.get("ssl.certificate.location").map(String::from);
79        self.ssl_key_location = config.get("ssl.key.location").map(String::from);
80        self.ssl_key_password = config.get("ssl.key.password").map(String::from);
81        Ok(())
82    }
83
84    fn parse_schema(&mut self, config: &ConnectorConfig) -> Result<(), ConnectorError> {
85        self.format = config.get("format").map_or(Ok(Format::Json), |format| {
86            format
87                .parse::<Format>()
88                .map_err(|error| ConnectorError::ConfigurationError(error.to_string()))
89        })?;
90        self.schema_registry_url = config.get("schema.registry.url").map(String::from);
91        self.schema_registry_auth = parse_schema_registry_auth(config)?;
92        self.schema_compatibility = config
93            .get("schema.compatibility")
94            .map(str::parse)
95            .transpose()?;
96        self.schema_evolution_strategy = config
97            .get("schema.evolution.strategy")
98            .map_or(Ok(SchemaEvolutionStrategy::default()), str::parse)?;
99        self.schema_registry_ssl_ca_location = config
100            .get("schema.registry.ssl.ca.location")
101            .map(String::from);
102        self.schema_registry_ssl_certificate_location = config
103            .get("schema.registry.ssl.certificate.location")
104            .map(String::from);
105        self.schema_registry_ssl_key_location = config
106            .get("schema.registry.ssl.key.location")
107            .map(String::from);
108        self.schema_registry_subject_strategy = config
109            .get("schema.registry.subject.name.strategy")
110            .map_or(Ok(SubjectNameStrategy::default()), str::parse)?;
111        self.schema_registry_record_name =
112            config.get("schema.registry.record.name").map(String::from);
113        validate_subject_name_strategy(
114            self.schema_registry_subject_strategy,
115            self.schema_registry_record_name.as_deref(),
116        )?;
117        self.schema_registry_discovery_timeout = config
118            .get_parsed::<u64>("schema.registry.discovery.timeout.ms")?
119            .map_or(Duration::from_secs(10), Duration::from_millis);
120        if let Some(key) = ["include.metadata", "include.headers"]
121            .into_iter()
122            .find(|key| config.get(key).is_some())
123        {
124            return Err(ConnectorError::ConfigurationError(format!(
125                "'{key}' was replaced by metadata.columns (key, topic, partition, offset, \
126                 timestamp, timestamp_type, headers)"
127            )));
128        }
129        self.metadata = config.get("metadata.columns").map_or(
130            Ok(KafkaMetadataColumns::default()),
131            KafkaMetadataColumns::parse,
132        )?;
133        Ok(())
134    }
135
136    fn parse_startup_and_fetch(&mut self, config: &ConnectorConfig) -> Result<(), ConnectorError> {
137        self.startup_mode = parse_startup_mode(config)?;
138        self.auto_offset_reset = match &self.startup_mode {
139            StartupMode::Earliest => OffsetReset::Earliest,
140            StartupMode::Latest => OffsetReset::Latest,
141            _ => config
142                .get("auto.offset.reset")
143                .map_or(Ok(OffsetReset::Earliest), str::parse)?,
144        };
145        self.isolation_level = config
146            .get("isolation.level")
147            .map_or(Ok(IsolationLevel::default()), str::parse)?;
148        self.latency_profile = config
149            .get("latency.profile")
150            .map_or(Ok(LatencyProfile::default()), str::parse)?;
151        let profile = self.latency_profile.source_defaults();
152        self.max_poll_records = config
153            .get_parsed::<usize>("max.poll.records")?
154            .unwrap_or(profile.max_poll_records);
155        self.partition_assignment_strategy = config
156            .get("partition.assignment.strategy")
157            .map_or(Ok(AssignmentStrategy::Range), str::parse)?;
158        self.fetch_min_bytes = config
159            .get_parsed::<i32>("fetch.min.bytes")?
160            .or(profile.fetch_min_bytes);
161        self.fetch_max_bytes = config.get_parsed::<i32>("fetch.max.bytes")?;
162        self.fetch_max_wait_ms = config
163            .get_parsed::<i32>("fetch.max.wait.ms")?
164            .or(profile.fetch_max_wait_ms);
165        self.max_partition_fetch_bytes = config.get_parsed::<i32>("max.partition.fetch.bytes")?;
166        Ok(())
167    }
168
169    fn parse_group_timing(&mut self, config: &ConnectorConfig) -> Result<(), ConnectorError> {
170        self.group_protocol = config
171            .get("group.protocol")
172            .map_or(Ok(GroupProtocol::default()), str::parse)?;
173        if self.group_protocol == GroupProtocol::Consumer {
174            if let Some(key) = CLASSIC_GROUP_OPTIONS
175                .iter()
176                .find(|key| config.get(key).is_some())
177            {
178                return Err(ConnectorError::ConfigurationError(format!(
179                    "'{key}' is a classic consumer-group setting; group.protocol = consumer \
180                     (KIP-848) assigns partitions and times sessions on the broker"
181                )));
182            }
183        }
184        let session_timeout_ms = config
185            .get_parsed::<u64>("session.timeout.ms")?
186            .unwrap_or(45_000);
187        let heartbeat_interval_ms = config
188            .get_parsed::<u64>("heartbeat.interval.ms")?
189            .unwrap_or(10_000);
190        self.queued_max_messages_kbytes = config
191            .get_parsed::<u32>("queued.max.messages.kbytes")?
192            .unwrap_or(16384);
193        let max_poll_interval_ms = config
194            .get_parsed::<u64>("max.poll.interval.ms")?
195            .unwrap_or(600_000);
196
197        self.session_timeout = Duration::from_millis(session_timeout_ms);
198        self.heartbeat_interval = Duration::from_millis(heartbeat_interval_ms);
199        self.max_poll_interval = Duration::from_millis(max_poll_interval_ms);
200        Ok(())
201    }
202
203    fn parse_delivery_controls(&mut self, config: &ConnectorConfig) -> Result<(), ConnectorError> {
204        reject_deprecated_commit_interval(config)?;
205        self.broker_commit_on_checkpoint = config
206            .get_parsed::<bool>("broker.commit.on.checkpoint")?
207            .unwrap_or(true);
208        self.reader_channel_capacity = config
209            .get_parsed::<usize>("reader.channel.capacity")?
210            .unwrap_or(8192);
211        if let Some(bytes) = config.get_parsed::<usize>("reader.channel.max.bytes")? {
212            self.reader_channel_max_bytes = bytes;
213        }
214        if let Some(bytes) = config.get_parsed::<usize>("max.poll.bytes")? {
215            self.max_poll_bytes = bytes;
216        }
217        if let Some(bytes) = config.get_parsed::<usize>("record.max.bytes")? {
218            self.record_max_bytes = bytes;
219        }
220        self.backpressure_high_watermark = config
221            .get_parsed::<f64>("backpressure.high.watermark")?
222            .unwrap_or(0.8);
223        self.backpressure_low_watermark = config
224            .get_parsed::<f64>("backpressure.low.watermark")?
225            .unwrap_or(0.25);
226        self.max_deser_error_rate = config
227            .get_parsed::<f64>("max.deser.error.rate")?
228            .unwrap_or(0.5);
229        Ok(())
230    }
231}
232
233fn parse_schema_registry_auth(config: &ConnectorConfig) -> Result<Option<SrAuth>, ConnectorError> {
234    match (
235        config.get("schema.registry.username"),
236        config.get("schema.registry.password"),
237    ) {
238        (Some(username), Some(password)) => Ok(Some(SrAuth {
239            username: username.to_string(),
240            password: password.to_string(),
241        })),
242        (Some(_), None) | (None, Some(_)) => Err(ConnectorError::ConfigurationError(
243            "schema.registry.username and schema.registry.password must both be set".to_string(),
244        )),
245        (None, None) => Ok(None),
246    }
247}
248
249fn validate_subject_name_strategy(
250    strategy: SubjectNameStrategy,
251    record_name: Option<&str>,
252) -> Result<(), ConnectorError> {
253    if matches!(
254        strategy,
255        SubjectNameStrategy::RecordName | SubjectNameStrategy::TopicRecordName
256    ) && record_name.is_none()
257    {
258        return Err(ConnectorError::ConfigurationError(format!(
259            "schema.registry.subject.name.strategy={strategy} requires schema.registry.record.name"
260        )));
261    }
262    Ok(())
263}
264
265fn parse_startup_mode(config: &ConnectorConfig) -> Result<StartupMode, ConnectorError> {
266    if let Some(offsets) = config.get("startup.specific.offsets") {
267        return Ok(StartupMode::SpecificOffsets(parse_specific_offsets(
268            offsets,
269        )?));
270    }
271    if let Some(timestamp) = config.get("startup.timestamp.ms") {
272        let timestamp = timestamp.parse().map_err(|_| {
273            ConnectorError::ConfigurationError(format!(
274                "invalid startup.timestamp.ms: '{timestamp}'"
275            ))
276        })?;
277        return Ok(StartupMode::Timestamp(timestamp));
278    }
279    config
280        .get("startup.mode")
281        .map_or(Ok(StartupMode::default()), str::parse)
282}
283
284fn reject_deprecated_commit_interval(config: &ConnectorConfig) -> Result<(), ConnectorError> {
285    if config
286        .properties()
287        .contains_key("broker.commit.interval.ms")
288    {
289        return Err(ConnectorError::ConfigurationError(
290            "broker.commit.interval.ms is no longer supported — broker offset commits now happen \
291             on checkpoint completion. Set broker.commit.on.checkpoint=false to disable."
292                .into(),
293        ));
294    }
295    Ok(())
296}
laminar_connectors/kafka/config/source_options.rs

laminar_connectors/kafka/config/
source_options.rs
