laminar_connectors/kafka/
sink_config.rs

1//! Kafka sink connector configuration.
2//!
3//! [`KafkaSinkConfig`] encapsulates all tuning knobs for the Kafka producer,
4//! parsed from a SQL `WITH (...)` clause via [`ConnectorConfig`].
5
6use std::collections::HashMap;
7use std::time::Duration;
8
9use rdkafka::ClientConfig;
10
11use crate::config::ConnectorConfig;
12use crate::error::ConnectorError;
13use crate::kafka::config::{CompatibilityLevel, SaslMechanism, SecurityProtocol, SrAuth};
14use crate::serde::Format;
15
16const MAX_DELIVERY_TIMEOUT: Duration = Duration::from_secs(300);
17
18/// Configuration for the Kafka Sink Connector.
19///
20/// Parsed from SQL `WITH (...)` clause options.
21///
22/// Uses a custom `Debug` impl that redacts `sasl_password` and
23/// `ssl_key_password` to prevent credential leakage in logs.
24#[derive(Clone)]
25pub struct KafkaSinkConfig {
26    /// Kafka broker addresses (comma-separated).
27    pub bootstrap_servers: String,
28    /// Target Kafka topic name.
29    pub topic: String,
30    /// Security protocol for broker connections.
31    pub security_protocol: SecurityProtocol,
32    /// SASL authentication mechanism.
33    pub sasl_mechanism: Option<SaslMechanism>,
34    /// SASL username (for PLAIN, SCRAM-SHA-256, SCRAM-SHA-512).
35    pub sasl_username: Option<String>,
36    /// SASL password (for PLAIN, SCRAM-SHA-256, SCRAM-SHA-512).
37    pub sasl_password: Option<String>,
38    /// Path to SSL CA certificate file (PEM format).
39    pub ssl_ca_location: Option<String>,
40    /// Path to client SSL certificate file (PEM format).
41    pub ssl_certificate_location: Option<String>,
42    /// Path to client SSL private key file (PEM format).
43    pub ssl_key_location: Option<String>,
44    /// Password for encrypted SSL private key.
45    pub ssl_key_password: Option<String>,
46    /// Serialization format.
47    pub format: Format,
48    /// Schema Registry URL for Avro.
49    pub schema_registry_url: Option<String>,
50    /// Schema Registry authentication.
51    pub schema_registry_auth: Option<SrAuth>,
52    /// Schema compatibility level override.
53    pub schema_compatibility: Option<CompatibilityLevel>,
54    /// Schema Registry SSL CA certificate path.
55    pub schema_registry_ssl_ca_location: Option<String>,
56    /// Maximum number of in-flight requests per connection.
57    pub max_in_flight: usize,
58    /// Maximum time to wait for delivery confirmation.
59    pub delivery_timeout: Duration,
60    /// Key column name for partitioning. In `envelope = upsert` mode this is the merge key
61    /// (the group identity), and is required.
62    pub key_column: Option<String>,
63    /// How an updating (changelog) input is encoded to the topic (`append` vs `upsert`).
64    pub envelope: SinkEnvelope,
65    /// Partitioning strategy.
66    pub partitioner: PartitionStrategy,
67    /// Maximum time to wait before sending a batch (milliseconds).
68    pub linger_ms: u64,
69    /// Maximum batch size in bytes.
70    pub batch_size: usize,
71    /// Maximum number of messages per batch.
72    pub batch_num_messages: Option<usize>,
73    /// Compression algorithm.
74    pub compression: CompressionType,
75    /// Dead letter queue topic for failed records.
76    pub dlq_topic: Option<String>,
77    /// Additional rdkafka client properties (pass-through).
78    pub kafka_properties: HashMap<String, String>,
79}
80
81/// How the sink encodes an updating (changelog) input to Kafka.
82#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
83pub enum SinkEnvelope {
84    /// Append-only: rows are produced as-is. A changelog's retractions are stripped upstream
85    /// (`prepare_for_sink`), so this cannot faithfully carry an aggregate's deletes.
86    #[default]
87    Append,
88    /// Upsert: the Z-set changelog is collapsed per key each batch — a live group becomes a keyed
89    /// record, a removed group a null-value tombstone. Consume via a log-compacted topic.
90    Upsert,
91}
92
93impl std::fmt::Debug for KafkaSinkConfig {
94    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
95        f.debug_struct("KafkaSinkConfig")
96            .field("bootstrap_servers", &self.bootstrap_servers)
97            .field("topic", &self.topic)
98            .field("format", &self.format)
99            .field("security_protocol", &self.security_protocol)
100            .field("sasl_mechanism", &self.sasl_mechanism)
101            .field("sasl_password", &self.sasl_password.as_ref().map(|_| "***"))
102            .field(
103                "ssl_key_password",
104                &self.ssl_key_password.as_ref().map(|_| "***"),
105            )
106            .field("partitioner", &self.partitioner)
107            .finish_non_exhaustive()
108    }
109}
110
111impl Default for KafkaSinkConfig {
112    fn default() -> Self {
113        Self {
114            bootstrap_servers: String::new(),
115            topic: String::new(),
116            security_protocol: SecurityProtocol::default(),
117            sasl_mechanism: None,
118            sasl_username: None,
119            sasl_password: None,
120            ssl_ca_location: None,
121            ssl_certificate_location: None,
122            ssl_key_location: None,
123            ssl_key_password: None,
124            format: Format::Json,
125            schema_registry_url: None,
126            schema_registry_auth: None,
127            schema_compatibility: None,
128            schema_registry_ssl_ca_location: None,
129            max_in_flight: 5,
130            delivery_timeout: Duration::from_secs(120),
131            key_column: None,
132            envelope: SinkEnvelope::default(),
133            partitioner: PartitionStrategy::KeyHash,
134            linger_ms: 5,
135            batch_size: 16_384,
136            batch_num_messages: None,
137            compression: CompressionType::None,
138            dlq_topic: None,
139            kafka_properties: HashMap::new(),
140        }
141    }
142}
143
144impl KafkaSinkConfig {
145    /// Parses a sink config from a [`ConnectorConfig`] (SQL WITH clause).
146    ///
147    /// # Errors
148    ///
149    /// Returns `ConnectorError::MissingConfig` if required keys are absent,
150    /// or `ConnectorError::ConfigurationError` on invalid values.
151    #[allow(clippy::too_many_lines, clippy::field_reassign_with_default)]
152    pub fn from_config(config: &ConnectorConfig) -> Result<Self, ConnectorError> {
153        let mut cfg = Self::default();
154
155        cfg.bootstrap_servers = config
156            .get("bootstrap.servers")
157            .ok_or_else(|| ConnectorError::missing_config("bootstrap.servers"))?
158            .to_string();
159
160        cfg.topic = config
161            .get("topic")
162            .ok_or_else(|| ConnectorError::missing_config("topic"))?
163            .to_string();
164
165        if let Some(s) = config.get("security.protocol") {
166            cfg.security_protocol = s.parse()?;
167        }
168
169        if let Some(s) = config.get("sasl.mechanism") {
170            cfg.sasl_mechanism = Some(s.parse()?);
171        }
172
173        cfg.sasl_username = config.get("sasl.username").map(String::from);
174        cfg.sasl_password = config.get("sasl.password").map(String::from);
175        cfg.ssl_ca_location = config.get("ssl.ca.location").map(String::from);
176        cfg.ssl_certificate_location = config.get("ssl.certificate.location").map(String::from);
177        cfg.ssl_key_location = config.get("ssl.key.location").map(String::from);
178        cfg.ssl_key_password = config.get("ssl.key.password").map(String::from);
179
180        if let Some(fmt) = config.get("format") {
181            cfg.format = fmt.parse().map_err(ConnectorError::Serde)?;
182        }
183
184        cfg.schema_registry_url = config.get("schema.registry.url").map(String::from);
185
186        let sr_user = config.get("schema.registry.username");
187        let sr_pass = config.get("schema.registry.password");
188        if let (Some(user), Some(pass)) = (sr_user, sr_pass) {
189            cfg.schema_registry_auth = Some(SrAuth {
190                username: user.to_string(),
191                password: pass.to_string(),
192            });
193        }
194
195        if let Some(c) = config.get("schema.compatibility") {
196            cfg.schema_compatibility = Some(c.parse().map_err(|_| {
197                ConnectorError::ConfigurationError(format!("invalid schema.compatibility: '{c}'"))
198            })?);
199        }
200
201        cfg.schema_registry_ssl_ca_location = config
202            .get("schema.registry.ssl.ca.location")
203            .map(String::from);
204
205        if config.get("acks").is_some() {
206            return Err(ConnectorError::ConfigurationError(
207                "'acks' is not configurable; the durable Kafka sink always requires acks=all"
208                    .into(),
209            ));
210        }
211
212        if let Some(v) = config.get("max.in.flight.requests") {
213            cfg.max_in_flight = v.parse().map_err(|_| {
214                ConnectorError::ConfigurationError(format!("invalid max.in.flight.requests: '{v}'"))
215            })?;
216        }
217
218        if let Some(v) = config.get("delivery.timeout.ms") {
219            let ms: u64 = v.parse().map_err(|_| {
220                ConnectorError::ConfigurationError(format!("invalid delivery.timeout.ms: '{v}'"))
221            })?;
222            cfg.delivery_timeout = Duration::from_millis(ms);
223        }
224
225        cfg.key_column = config.get("key.column").map(String::from);
226
227        cfg.envelope = match config
228            .get("envelope")
229            .map(str::to_ascii_lowercase)
230            .as_deref()
231        {
232            None | Some("append") => SinkEnvelope::Append,
233            Some("upsert") => SinkEnvelope::Upsert,
234            Some(other) => {
235                return Err(ConnectorError::ConfigurationError(format!(
236                    "unknown envelope '{other}' (expected 'append' or 'upsert')"
237                )));
238            }
239        };
240        if cfg.envelope == SinkEnvelope::Upsert
241            && cfg
242                .key_column
243                .as_deref()
244                .is_none_or(|k| k.trim().is_empty())
245        {
246            return Err(ConnectorError::ConfigurationError(
247                "envelope = 'upsert' requires a non-empty 'key.column' (the merge key)".into(),
248            ));
249        }
250
251        if let Some(p) = config.get("partitioner") {
252            cfg.partitioner = p.parse().map_err(|_| {
253                ConnectorError::ConfigurationError(format!(
254                    "invalid partitioner: '{p}' (expected 'key-hash', 'round-robin', or 'sticky')"
255                ))
256            })?;
257        }
258
259        if let Some(v) = config.get("linger.ms") {
260            cfg.linger_ms = v.parse().map_err(|_| {
261                ConnectorError::ConfigurationError(format!("invalid linger.ms: '{v}'"))
262            })?;
263        }
264
265        if let Some(v) = config.get("batch.size") {
266            cfg.batch_size = v.parse().map_err(|_| {
267                ConnectorError::ConfigurationError(format!("invalid batch.size: '{v}'"))
268            })?;
269        }
270
271        if let Some(v) = config.get("batch.num.messages") {
272            cfg.batch_num_messages = Some(v.parse().map_err(|_| {
273                ConnectorError::ConfigurationError(format!("invalid batch.num.messages: '{v}'"))
274            })?);
275        }
276
277        if let Some(c) = config.get("compression.type") {
278            cfg.compression = c.parse().map_err(|_| {
279                ConnectorError::ConfigurationError(format!("invalid compression.type: '{c}'"))
280            })?;
281        }
282
283        cfg.dlq_topic = config.get("dlq.topic").map(String::from);
284        if cfg.dlq_topic.is_some() && cfg.envelope == SinkEnvelope::Upsert {
285            // Upsert failures poison the epoch; a lone tombstone in a DLQ would corrupt the
286            // compacted topic. Reject rather than silently ignore the DLQ.
287            return Err(ConnectorError::ConfigurationError(
288                "'dlq.topic' is not supported with envelope = 'upsert' (upsert failures poison the \
289                 epoch instead of routing to a DLQ)"
290                    .into(),
291            ));
292        }
293
294        for (key, value) in config.properties_with_prefix("kafka.") {
295            cfg.kafka_properties.insert(key, value);
296        }
297
298        cfg.validate()?;
299        Ok(cfg)
300    }
301
302    /// Validates the configuration.
303    ///
304    /// # Errors
305    ///
306    /// Returns `ConnectorError::ConfigurationError` on invalid combinations.
307    pub fn validate(&self) -> Result<(), ConnectorError> {
308        if self.bootstrap_servers.is_empty() {
309            return Err(ConnectorError::missing_config("bootstrap.servers"));
310        }
311        if self.topic.is_empty() {
312            return Err(ConnectorError::missing_config("topic"));
313        }
314
315        if self.security_protocol.uses_sasl() && self.sasl_mechanism.is_none() {
316            return Err(ConnectorError::ConfigurationError(
317                "sasl.mechanism is required when security.protocol is sasl_plaintext or sasl_ssl"
318                    .into(),
319            ));
320        }
321
322        if let Some(mechanism) = &self.sasl_mechanism {
323            if mechanism.requires_credentials()
324                && (self.sasl_username.is_none() || self.sasl_password.is_none())
325            {
326                return Err(ConnectorError::ConfigurationError(format!(
327                    "sasl.username and sasl.password are required for {mechanism} mechanism"
328                )));
329            }
330        }
331
332        if self.security_protocol.uses_ssl() {
333            if let Some(ref ca) = self.ssl_ca_location {
334                if ca.is_empty() {
335                    return Err(ConnectorError::ConfigurationError(
336                        "ssl.ca.location cannot be empty when specified".into(),
337                    ));
338                }
339            }
340        }
341
342        if self.format == Format::Debezium {
343            return Err(ConnectorError::ConfigurationError(
344                "Debezium is a deserialization-only format and cannot be used for sinks".into(),
345            ));
346        }
347
348        if self.format == Format::Avro && self.schema_registry_url.is_none() {
349            return Err(ConnectorError::ConfigurationError(
350                "Avro format requires 'schema.registry.url'".into(),
351            ));
352        }
353
354        if !(1..=5).contains(&self.max_in_flight) {
355            return Err(ConnectorError::ConfigurationError(
356                "max.in.flight.requests must be between 1 and 5 when idempotence is enabled".into(),
357            ));
358        }
359
360        if self.delivery_timeout.is_zero() {
361            return Err(ConnectorError::ConfigurationError(
362                "delivery.timeout.ms must be > 0; zero disables librdkafka's delivery deadline"
363                    .into(),
364            ));
365        }
366        if self.delivery_timeout > MAX_DELIVERY_TIMEOUT {
367            return Err(ConnectorError::ConfigurationError(format!(
368                "delivery.timeout.ms must be <= {}",
369                MAX_DELIVERY_TIMEOUT.as_millis()
370            )));
371        }
372
373        Ok(())
374    }
375
376    /// Builds an rdkafka [`ClientConfig`] from this configuration.
377    ///
378    /// Always sets `enable.idempotence=true` for durable at-least-once writes.
379    #[must_use]
380    pub fn to_rdkafka_config(&self) -> ClientConfig {
381        let mut config = ClientConfig::new();
382
383        config.set("bootstrap.servers", &self.bootstrap_servers);
384        config.set("security.protocol", self.security_protocol.as_rdkafka_str());
385
386        if let Some(ref mechanism) = self.sasl_mechanism {
387            config.set("sasl.mechanism", mechanism.as_rdkafka_str());
388        }
389
390        if let Some(ref username) = self.sasl_username {
391            config.set("sasl.username", username);
392        }
393
394        if let Some(ref password) = self.sasl_password {
395            config.set("sasl.password", password);
396        }
397
398        if let Some(ref ca) = self.ssl_ca_location {
399            config.set("ssl.ca.location", ca);
400        }
401
402        if let Some(ref cert) = self.ssl_certificate_location {
403            config.set("ssl.certificate.location", cert);
404        }
405
406        if let Some(ref key) = self.ssl_key_location {
407            config.set("ssl.key.location", key);
408        }
409
410        if let Some(ref key_pass) = self.ssl_key_password {
411            config.set("ssl.key.password", key_pass);
412        }
413
414        config
415            .set("enable.idempotence", "true")
416            .set("acks", "all")
417            .set("linger.ms", self.linger_ms.to_string())
418            .set("batch.size", self.batch_size.to_string())
419            .set("compression.type", self.compression.as_rdkafka_str())
420            .set(
421                "max.in.flight.requests.per.connection",
422                self.max_in_flight.to_string(),
423            )
424            .set(
425                "message.timeout.ms",
426                self.delivery_timeout.as_millis().to_string(),
427            );
428
429        if let Some(num_msgs) = self.batch_num_messages {
430            config.set("batch.num.messages", num_msgs.to_string());
431        }
432
433        // Apply pass-through properties, blocking security-critical keys
434        // that could silently downgrade authentication or break semantics.
435        for (key, value) in &self.kafka_properties {
436            if is_blocked_passthrough_key(key) {
437                tracing::warn!(
438                    key,
439                    "ignoring kafka.* pass-through property that overrides a protected connector setting"
440                );
441                continue;
442            }
443            config.set(key, value);
444        }
445
446        config
447    }
448
449    /// Builds an rdkafka [`ClientConfig`] for the dead letter queue producer.
450    ///
451    /// Inherits security settings (SASL, SSL) from the main config but is
452    /// non-transactional. Does not set `transactional.id`.
453    #[must_use]
454    pub fn to_dlq_rdkafka_config(&self) -> ClientConfig {
455        let mut config = ClientConfig::new();
456
457        config.set("bootstrap.servers", &self.bootstrap_servers);
458        config.set("security.protocol", self.security_protocol.as_rdkafka_str());
459
460        if let Some(ref mechanism) = self.sasl_mechanism {
461            config.set("sasl.mechanism", mechanism.as_rdkafka_str());
462        }
463        if let Some(ref username) = self.sasl_username {
464            config.set("sasl.username", username);
465        }
466        if let Some(ref password) = self.sasl_password {
467            config.set("sasl.password", password);
468        }
469        if let Some(ref ca) = self.ssl_ca_location {
470            config.set("ssl.ca.location", ca);
471        }
472        if let Some(ref cert) = self.ssl_certificate_location {
473            config.set("ssl.certificate.location", cert);
474        }
475        if let Some(ref key) = self.ssl_key_location {
476            config.set("ssl.key.location", key);
477        }
478        if let Some(ref key_pass) = self.ssl_key_password {
479            config.set("ssl.key.password", key_pass);
480        }
481
482        config
483            .set("enable.idempotence", "true")
484            .set("acks", "all")
485            .set(
486                "max.in.flight.requests.per.connection",
487                self.max_in_flight.to_string(),
488            )
489            .set(
490                "message.timeout.ms",
491                self.delivery_timeout.as_millis().to_string(),
492            );
493
494        config
495    }
496}
497
498/// Returns `true` if a pass-through kafka.* key must not override explicit settings.
499fn is_blocked_passthrough_key(key: &str) -> bool {
500    key.starts_with("sasl.kerberos.")
501        || matches!(
502            key,
503            "security.protocol"
504                | "sasl.mechanism"
505                | "sasl.username"
506                | "sasl.password"
507                | "sasl.oauthbearer.config"
508                | "ssl.ca.location"
509                | "ssl.certificate.location"
510                | "ssl.key.location"
511                | "ssl.key.password"
512                | "ssl.endpoint.identification.algorithm"
513                | "enable.auto.commit"
514                | "enable.idempotence"
515                | "acks"
516                | "max.in.flight"
517                | "max.in.flight.requests.per.connection"
518                | "message.timeout.ms"
519                | "delivery.timeout.ms"
520                | "transactional.id"
521        )
522}
523
524/// Partitioning strategy for distributing records across Kafka partitions.
525#[derive(Debug, Clone, Copy, PartialEq, Eq)]
526pub enum PartitionStrategy {
527    /// Hash the key column (Murmur2, Kafka-compatible).
528    KeyHash,
529    /// Round-robin across all partitions.
530    RoundRobin,
531    /// Sticky: batch records to the same partition until full.
532    Sticky,
533}
534
535str_enum!(PartitionStrategy, lowercase_udash, String, "unknown partition strategy",
536    KeyHash => "key-hash", "keyhash", "hash";
537    RoundRobin => "round-robin", "roundrobin";
538    Sticky => "sticky"
539);
540
541/// Compression type for produced Kafka messages.
542#[derive(Debug, Clone, Copy, PartialEq, Eq)]
543pub enum CompressionType {
544    /// No compression.
545    None,
546    /// Gzip compression.
547    Gzip,
548    /// Snappy compression.
549    Snappy,
550    /// LZ4 compression.
551    Lz4,
552    /// Zstandard compression.
553    Zstd,
554}
555
556impl CompressionType {
557    /// Returns the rdkafka configuration string.
558    #[must_use]
559    pub fn as_rdkafka_str(&self) -> &'static str {
560        match self {
561            Self::None => "none",
562            Self::Gzip => "gzip",
563            Self::Snappy => "snappy",
564            Self::Lz4 => "lz4",
565            Self::Zstd => "zstd",
566        }
567    }
568}
569
570str_enum!(fromstr CompressionType, lowercase_nodash, String, "unknown compression type",
571    None => "none";
572    Gzip => "gzip";
573    Snappy => "snappy";
574    Lz4 => "lz4";
575    Zstd => "zstd", "zstandard"
576);
577
578impl std::fmt::Display for CompressionType {
579    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
580        write!(f, "{}", self.as_rdkafka_str())
581    }
582}
583
584#[cfg(test)]
585mod tests {
586    use super::*;
587
588    fn make_config(pairs: &[(&str, &str)]) -> ConnectorConfig {
589        let mut config = ConnectorConfig::new("kafka");
590        for (k, v) in pairs {
591            config.set(*k, *v);
592        }
593        config
594    }
595
596    fn required_pairs() -> Vec<(&'static str, &'static str)> {
597        vec![
598            ("bootstrap.servers", "localhost:9092"),
599            ("topic", "output-events"),
600        ]
601    }
602
603    #[test]
604    fn test_parse_required_fields() {
605        let config = make_config(&required_pairs());
606        let cfg = KafkaSinkConfig::from_config(&config).unwrap();
607        assert_eq!(cfg.bootstrap_servers, "localhost:9092");
608        assert_eq!(cfg.topic, "output-events");
609        assert_eq!(cfg.format, Format::Json);
610        assert_eq!(cfg.security_protocol, SecurityProtocol::Plaintext);
611    }
612
613    #[test]
614    fn test_missing_bootstrap_servers() {
615        let config = make_config(&[("topic", "t")]);
616        assert!(KafkaSinkConfig::from_config(&config).is_err());
617    }
618
619    #[test]
620    fn test_missing_topic() {
621        let config = make_config(&[("bootstrap.servers", "b:9092")]);
622        assert!(KafkaSinkConfig::from_config(&config).is_err());
623    }
624
625    #[test]
626    fn test_parse_security_sasl_ssl() {
627        let mut pairs = required_pairs();
628        pairs.extend_from_slice(&[
629            ("security.protocol", "sasl_ssl"),
630            ("sasl.mechanism", "SCRAM-SHA-512"),
631            ("sasl.username", "producer"),
632            ("sasl.password", "secret123"),
633            ("ssl.ca.location", "/etc/ssl/ca.pem"),
634        ]);
635        let config = make_config(&pairs);
636        let cfg = KafkaSinkConfig::from_config(&config).unwrap();
637
638        assert_eq!(cfg.security_protocol, SecurityProtocol::SaslSsl);
639        assert_eq!(cfg.sasl_mechanism, Some(SaslMechanism::ScramSha512));
640        assert_eq!(cfg.sasl_username, Some("producer".to_string()));
641        assert_eq!(cfg.sasl_password, Some("secret123".to_string()));
642        assert_eq!(cfg.ssl_ca_location, Some("/etc/ssl/ca.pem".to_string()));
643    }
644
645    #[test]
646    fn test_parse_security_ssl_only() {
647        let mut pairs = required_pairs();
648        pairs.extend_from_slice(&[
649            ("security.protocol", "ssl"),
650            ("ssl.ca.location", "/etc/ssl/ca.pem"),
651            ("ssl.certificate.location", "/etc/ssl/client.pem"),
652            ("ssl.key.location", "/etc/ssl/client.key"),
653            ("ssl.key.password", "keypass"),
654        ]);
655        let config = make_config(&pairs);
656        let cfg = KafkaSinkConfig::from_config(&config).unwrap();
657
658        assert_eq!(cfg.security_protocol, SecurityProtocol::Ssl);
659        assert_eq!(cfg.ssl_ca_location, Some("/etc/ssl/ca.pem".to_string()));
660        assert_eq!(
661            cfg.ssl_certificate_location,
662            Some("/etc/ssl/client.pem".to_string())
663        );
664        assert_eq!(
665            cfg.ssl_key_location,
666            Some("/etc/ssl/client.key".to_string())
667        );
668        assert_eq!(cfg.ssl_key_password, Some("keypass".to_string()));
669    }
670
671    #[test]
672    fn test_parse_all_optional_fields() {
673        let mut pairs = required_pairs();
674        pairs.extend_from_slice(&[
675            ("format", "avro"),
676            ("key.column", "order_id"),
677            ("partitioner", "round-robin"),
678            ("linger.ms", "10"),
679            ("batch.size", "32768"),
680            ("batch.num.messages", "5000"),
681            ("compression.type", "zstd"),
682            ("max.in.flight.requests", "3"),
683            ("delivery.timeout.ms", "60000"),
684            ("dlq.topic", "my-dlq"),
685            ("schema.registry.url", "http://sr:8081"),
686            ("schema.registry.username", "user"),
687            ("schema.registry.password", "pass"),
688            ("schema.registry.ssl.ca.location", "/etc/ssl/sr-ca.pem"),
689        ]);
690        let config = make_config(&pairs);
691        let cfg = KafkaSinkConfig::from_config(&config).unwrap();
692
693        assert_eq!(cfg.format, Format::Avro);
694        assert_eq!(cfg.key_column.as_deref(), Some("order_id"));
695        assert_eq!(cfg.partitioner, PartitionStrategy::RoundRobin);
696        assert_eq!(cfg.linger_ms, 10);
697        assert_eq!(cfg.batch_size, 32_768);
698        assert_eq!(cfg.batch_num_messages, Some(5000));
699        assert_eq!(cfg.compression, CompressionType::Zstd);
700        assert_eq!(cfg.max_in_flight, 3);
701        assert_eq!(cfg.delivery_timeout, Duration::from_secs(60));
702        assert_eq!(cfg.dlq_topic.as_deref(), Some("my-dlq"));
703        assert_eq!(cfg.schema_registry_url.as_deref(), Some("http://sr:8081"));
704        assert!(cfg.schema_registry_auth.is_some());
705        assert_eq!(
706            cfg.schema_registry_ssl_ca_location,
707            Some("/etc/ssl/sr-ca.pem".to_string())
708        );
709    }
710
711    #[test]
712    fn delivery_timeout_rejects_infinite_or_excessive_values() {
713        for value in ["0", "300001"] {
714            let mut pairs = required_pairs();
715            pairs.push(("delivery.timeout.ms", value));
716            let error = KafkaSinkConfig::from_config(&make_config(&pairs)).unwrap_err();
717            assert!(
718                matches!(error, ConnectorError::ConfigurationError(_)),
719                "unexpected error for delivery.timeout.ms={value}: {error}"
720            );
721        }
722    }
723
724    #[test]
725    fn delivery_timeout_accepts_fixed_maximum() {
726        let mut pairs = required_pairs();
727        pairs.push(("delivery.timeout.ms", "300000"));
728        let cfg = KafkaSinkConfig::from_config(&make_config(&pairs)).unwrap();
729        assert_eq!(cfg.delivery_timeout, MAX_DELIVERY_TIMEOUT);
730        assert_eq!(
731            cfg.to_rdkafka_config().get("message.timeout.ms"),
732            Some("300000")
733        );
734        assert_eq!(
735            cfg.to_dlq_rdkafka_config().get("message.timeout.ms"),
736            Some("300000")
737        );
738    }
739
740    #[test]
741    fn test_validate_avro_requires_sr() {
742        let mut cfg = KafkaSinkConfig::default();
743        cfg.bootstrap_servers = "b:9092".into();
744        cfg.topic = "t".into();
745        cfg.format = Format::Avro;
746        assert!(cfg.validate().is_err());
747    }
748
749    #[test]
750    fn test_validate_sasl_without_mechanism() {
751        let mut cfg = KafkaSinkConfig::default();
752        cfg.bootstrap_servers = "b:9092".into();
753        cfg.topic = "t".into();
754        cfg.security_protocol = SecurityProtocol::SaslSsl;
755        // sasl_mechanism not set
756        assert!(cfg.validate().is_err());
757    }
758
759    #[test]
760    fn test_validate_sasl_plain_without_credentials() {
761        let mut cfg = KafkaSinkConfig::default();
762        cfg.bootstrap_servers = "b:9092".into();
763        cfg.topic = "t".into();
764        cfg.security_protocol = SecurityProtocol::SaslPlaintext;
765        cfg.sasl_mechanism = Some(SaslMechanism::ScramSha256);
766        // username/password not set
767        assert!(cfg.validate().is_err());
768    }
769
770    #[test]
771    fn test_rdkafka_config_at_least_once() {
772        let mut cfg = KafkaSinkConfig::default();
773        cfg.bootstrap_servers = "b:9092".into();
774        cfg.topic = "t".into();
775        let rdk = cfg.to_rdkafka_config();
776        assert_eq!(rdk.get("enable.idempotence"), Some("true"));
777        assert_eq!(rdk.get("acks"), Some("all"));
778        assert_eq!(rdk.get("max.in.flight.requests.per.connection"), Some("5"));
779        assert!(rdk.get("transactional.id").is_none());
780        assert_eq!(rdk.get("security.protocol"), Some("plaintext"));
781
782        let dlq = cfg.to_dlq_rdkafka_config();
783        assert_eq!(dlq.get("enable.idempotence"), Some("true"));
784        assert_eq!(dlq.get("acks"), Some("all"));
785        assert_eq!(dlq.get("max.in.flight.requests.per.connection"), Some("5"));
786    }
787
788    #[test]
789    fn acks_option_is_rejected_instead_of_downgrading_durability() {
790        for value in ["0", "1", "all"] {
791            let mut pairs = required_pairs();
792            pairs.push(("acks", value));
793            let error = KafkaSinkConfig::from_config(&make_config(&pairs)).unwrap_err();
794            assert!(matches!(error, ConnectorError::ConfigurationError(_)));
795            assert!(error.to_string().contains("always requires acks=all"));
796        }
797    }
798
799    #[test]
800    fn idempotent_producer_rejects_invalid_in_flight_boundaries() {
801        for value in ["0", "6", "100"] {
802            let mut pairs = required_pairs();
803            pairs.push(("max.in.flight.requests", value));
804            let error = KafkaSinkConfig::from_config(&make_config(&pairs)).unwrap_err();
805            assert!(matches!(error, ConnectorError::ConfigurationError(_)));
806        }
807        for value in ["1", "5"] {
808            let mut pairs = required_pairs();
809            pairs.push(("max.in.flight.requests", value));
810            assert!(KafkaSinkConfig::from_config(&make_config(&pairs)).is_ok());
811        }
812    }
813
814    #[test]
815    fn test_rdkafka_config_with_security() {
816        let mut cfg = KafkaSinkConfig::default();
817        cfg.bootstrap_servers = "b:9092".into();
818        cfg.topic = "t".into();
819        cfg.security_protocol = SecurityProtocol::SaslSsl;
820        cfg.sasl_mechanism = Some(SaslMechanism::Plain);
821        cfg.sasl_username = Some("user".into());
822        cfg.sasl_password = Some("pass".into());
823        cfg.ssl_ca_location = Some("/ca.pem".into());
824
825        let rdk = cfg.to_rdkafka_config();
826        assert_eq!(rdk.get("security.protocol"), Some("sasl_ssl"));
827        assert_eq!(rdk.get("sasl.mechanism"), Some("PLAIN"));
828        assert_eq!(rdk.get("sasl.username"), Some("user"));
829        assert_eq!(rdk.get("sasl.password"), Some("pass"));
830        assert_eq!(rdk.get("ssl.ca.location"), Some("/ca.pem"));
831    }
832
833    #[test]
834    fn test_rdkafka_config_with_batch_num_messages() {
835        let mut cfg = KafkaSinkConfig::default();
836        cfg.bootstrap_servers = "b:9092".into();
837        cfg.topic = "t".into();
838        cfg.batch_num_messages = Some(10_000);
839
840        let rdk = cfg.to_rdkafka_config();
841        assert_eq!(rdk.get("batch.num.messages"), Some("10000"));
842    }
843
844    #[test]
845    fn test_kafka_passthrough_properties() {
846        let mut pairs = required_pairs();
847        pairs.push(("kafka.socket.timeout.ms", "5000"));
848        pairs.push(("kafka.queue.buffering.max.messages", "100000"));
849        let config = make_config(&pairs);
850        let cfg = KafkaSinkConfig::from_config(&config).unwrap();
851        assert_eq!(
852            cfg.kafka_properties.get("socket.timeout.ms").unwrap(),
853            "5000"
854        );
855    }
856
857    #[test]
858    fn passthrough_cannot_disable_delivery_deadline() {
859        let mut pairs = required_pairs();
860        pairs.push(("kafka.message.timeout.ms", "0"));
861        pairs.push(("kafka.delivery.timeout.ms", "0"));
862        let config = make_config(&pairs);
863        let cfg = KafkaSinkConfig::from_config(&config).unwrap();
864
865        assert_eq!(
866            cfg.to_rdkafka_config().get("message.timeout.ms"),
867            Some("120000")
868        );
869        assert_eq!(
870            cfg.to_dlq_rdkafka_config().get("message.timeout.ms"),
871            Some("120000")
872        );
873    }
874
875    #[test]
876    fn passthrough_cannot_override_idempotent_delivery_invariants() {
877        let mut pairs = required_pairs();
878        pairs.push(("kafka.acks", "0"));
879        pairs.push(("kafka.max.in.flight", "99"));
880        pairs.push(("kafka.max.in.flight.requests.per.connection", "99"));
881        let cfg = KafkaSinkConfig::from_config(&make_config(&pairs)).unwrap();
882
883        for rdk in [cfg.to_rdkafka_config(), cfg.to_dlq_rdkafka_config()] {
884            assert_eq!(rdk.get("acks"), Some("all"));
885            assert_eq!(rdk.get("max.in.flight.requests.per.connection"), Some("5"));
886        }
887    }
888
889    #[test]
890    fn test_defaults() {
891        let cfg = KafkaSinkConfig::default();
892        assert_eq!(cfg.partitioner, PartitionStrategy::KeyHash);
893        assert_eq!(cfg.compression, CompressionType::None);
894        assert_eq!(cfg.linger_ms, 5);
895        assert_eq!(cfg.batch_size, 16_384);
896        assert_eq!(cfg.max_in_flight, 5);
897        assert_eq!(cfg.delivery_timeout, Duration::from_secs(120));
898        assert_eq!(cfg.security_protocol, SecurityProtocol::Plaintext);
899        assert!(cfg.sasl_mechanism.is_none());
900        assert!(cfg.batch_num_messages.is_none());
901    }
902
903    #[test]
904    fn test_enum_display() {
905        assert_eq!(PartitionStrategy::KeyHash.to_string(), "key-hash");
906        assert_eq!(PartitionStrategy::RoundRobin.to_string(), "round-robin");
907        assert_eq!(PartitionStrategy::Sticky.to_string(), "sticky");
908        assert_eq!(CompressionType::Zstd.to_string(), "zstd");
909    }
910
911    #[test]
912    fn test_enum_parse() {
913        assert_eq!(
914            "key-hash".parse::<PartitionStrategy>().unwrap(),
915            PartitionStrategy::KeyHash
916        );
917        assert_eq!(
918            "round-robin".parse::<PartitionStrategy>().unwrap(),
919            PartitionStrategy::RoundRobin
920        );
921        assert_eq!(
922            "sticky".parse::<PartitionStrategy>().unwrap(),
923            PartitionStrategy::Sticky
924        );
925        assert_eq!(
926            "gzip".parse::<CompressionType>().unwrap(),
927            CompressionType::Gzip
928        );
929        assert_eq!(
930            "snappy".parse::<CompressionType>().unwrap(),
931            CompressionType::Snappy
932        );
933        assert_eq!(
934            "lz4".parse::<CompressionType>().unwrap(),
935            CompressionType::Lz4
936        );
937        assert_eq!(
938            "zstd".parse::<CompressionType>().unwrap(),
939            CompressionType::Zstd
940        );
941    }
942}
laminar_connectors/kafka/sink_config.rs

laminar_connectors/kafka/
sink_config.rs
