laminar_connectors/kafka/sink_config/
mod.rs

1//! Kafka sink connector configuration.
2//!
3//! [`KafkaSinkConfig`] encapsulates all tuning knobs for the Kafka producer,
4//! parsed from the resolved sink [`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 resolved sink connector and format 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. Weighted changelogs are rejected during sink
85    /// admission and again at the runtime boundary because this envelope cannot carry retractions.
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 resolved [`ConnectorConfig`].
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;
laminar_connectors/kafka/sink_config/mod.rs

laminar_connectors/kafka/sink_config/
mod.rs
