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 arrow_schema::DataType;
10use rdkafka::ClientConfig;
11
12use crate::config::ConnectorConfig;
13use crate::connector::DeliveryGuarantee;
14use crate::error::ConnectorError;
15use crate::kafka::config::{CompatibilityLevel, SaslMechanism, SecurityProtocol, SrAuth};
16use crate::kafka::latency_profile::LatencyProfile;
17use crate::serde::Format;
18
19const MAX_DELIVERY_TIMEOUT: Duration = Duration::from_secs(300);
20/// Kafka's default broker `transaction.max.timeout.ms`.
21const MAX_TRANSACTION_TIMEOUT: Duration = Duration::from_secs(900);
22const DEFAULT_EPOCH_MAX_BYTES: usize = 64 * 1024 * 1024;
23/// Largest `epoch.max.bytes`; also the protocol bound on any staged descriptor's body.
24pub(crate) const MAX_EPOCH_BYTES: usize = 256 * 1024 * 1024;
25
26/// Configuration for the Kafka Sink Connector.
27///
28/// Parsed from resolved sink connector and format options.
29///
30/// Uses a custom `Debug` impl that redacts `sasl_password` and
31/// `ssl_key_password` to prevent credential leakage in logs.
32#[derive(Clone)]
33pub struct KafkaSinkConfig {
34    /// Kafka broker addresses (comma-separated).
35    pub bootstrap_servers: String,
36    /// Target Kafka topic name.
37    pub topic: String,
38    /// Security protocol for broker connections.
39    pub security_protocol: SecurityProtocol,
40    /// SASL authentication mechanism.
41    pub sasl_mechanism: Option<SaslMechanism>,
42    /// SASL username (for PLAIN, SCRAM-SHA-256, SCRAM-SHA-512).
43    pub sasl_username: Option<String>,
44    /// SASL password (for PLAIN, SCRAM-SHA-256, SCRAM-SHA-512).
45    pub sasl_password: Option<String>,
46    /// Path to SSL CA certificate file (PEM format).
47    pub ssl_ca_location: Option<String>,
48    /// Path to client SSL certificate file (PEM format).
49    pub ssl_certificate_location: Option<String>,
50    /// Path to client SSL private key file (PEM format).
51    pub ssl_key_location: Option<String>,
52    /// Password for encrypted SSL private key.
53    pub ssl_key_password: Option<String>,
54    /// Serialization format.
55    pub format: Format,
56    /// Schema Registry URL for Avro.
57    pub schema_registry_url: Option<String>,
58    /// Schema Registry authentication.
59    pub schema_registry_auth: Option<SrAuth>,
60    /// Schema compatibility level override.
61    pub schema_compatibility: Option<CompatibilityLevel>,
62    /// Schema Registry SSL CA certificate path.
63    pub schema_registry_ssl_ca_location: Option<String>,
64    /// Maximum number of in-flight requests per connection.
65    pub max_in_flight: usize,
66    /// Maximum time to wait for delivery confirmation.
67    pub delivery_timeout: Duration,
68    /// Comma-separated key column names. In `envelope = upsert` mode they are the merge key (the
69    /// group identity) and are required.
70    pub key_column: Option<String>,
71    /// How key column values become Kafka record-key bytes.
72    pub key_format: KeyFormat,
73    /// Partition count keyed records are hashed over. When set, the topic must have at least this
74    /// many partitions, and later partition additions do not move existing keys. Required for
75    /// `envelope = upsert`.
76    pub partition_count: Option<i32>,
77    /// How an updating (changelog) input is encoded to the topic (`append` vs `upsert`).
78    pub envelope: SinkEnvelope,
79    /// Partitioning strategy.
80    pub partitioner: PartitionStrategy,
81    /// Maximum time to wait before sending a batch (milliseconds).
82    pub linger_ms: u64,
83    /// Maximum batch size in bytes.
84    pub batch_size: usize,
85    /// Maximum number of messages per batch.
86    pub batch_num_messages: Option<usize>,
87    /// Compression algorithm.
88    pub compression: CompressionType,
89    /// Dead letter queue topic for failed records.
90    pub dlq_topic: Option<String>,
91    /// Latency preset that supplied the batching defaults.
92    pub latency_profile: LatencyProfile,
93    /// Pipeline delivery guarantee, injected by the runtime.
94    pub delivery: DeliveryGuarantee,
95    /// Compacted topic holding one commit marker per exactly-once checkpoint (exactly-once only).
96    pub commit_topic: Option<String>,
97    /// Broker transaction timeout for exactly-once publication.
98    pub transaction_timeout: Duration,
99    /// Bound on one exactly-once epoch's staged key and value bytes.
100    pub epoch_max_bytes: usize,
101    /// Additional rdkafka client properties (pass-through).
102    pub kafka_properties: HashMap<String, String>,
103}
104
105/// How the sink encodes an updating (changelog) input to Kafka.
106#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
107pub enum SinkEnvelope {
108    /// Append-only: rows are produced as-is. Weighted changelogs are rejected during sink
109    /// admission and again at the runtime boundary because this envelope cannot carry retractions.
110    #[default]
111    Append,
112    /// Upsert: the Z-set changelog is collapsed per key each batch — a live group becomes a keyed
113    /// record, a removed group a null-value tombstone. Consume via a log-compacted topic.
114    Upsert,
115}
116
117impl std::fmt::Debug for KafkaSinkConfig {
118    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
119        f.debug_struct("KafkaSinkConfig")
120            .field("bootstrap_servers", &self.bootstrap_servers)
121            .field("topic", &self.topic)
122            .field("format", &self.format)
123            .field("security_protocol", &self.security_protocol)
124            .field("sasl_mechanism", &self.sasl_mechanism)
125            .field("sasl_password", &self.sasl_password.as_ref().map(|_| "***"))
126            .field(
127                "ssl_key_password",
128                &self.ssl_key_password.as_ref().map(|_| "***"),
129            )
130            .field("partitioner", &self.partitioner)
131            .finish_non_exhaustive()
132    }
133}
134
135impl Default for KafkaSinkConfig {
136    fn default() -> Self {
137        Self {
138            bootstrap_servers: String::new(),
139            topic: String::new(),
140            security_protocol: SecurityProtocol::default(),
141            sasl_mechanism: None,
142            sasl_username: None,
143            sasl_password: None,
144            ssl_ca_location: None,
145            ssl_certificate_location: None,
146            ssl_key_location: None,
147            ssl_key_password: None,
148            format: Format::Json,
149            schema_registry_url: None,
150            schema_registry_auth: None,
151            schema_compatibility: None,
152            schema_registry_ssl_ca_location: None,
153            max_in_flight: 5,
154            delivery_timeout: Duration::from_secs(120),
155            key_column: None,
156            key_format: KeyFormat::default(),
157            partition_count: None,
158            envelope: SinkEnvelope::default(),
159            partitioner: PartitionStrategy::KeyHash,
160            linger_ms: 5,
161            batch_size: 16_384,
162            batch_num_messages: None,
163            compression: CompressionType::None,
164            dlq_topic: None,
165            latency_profile: LatencyProfile::Balanced,
166            delivery: DeliveryGuarantee::AtLeastOnce,
167            commit_topic: None,
168            transaction_timeout: Duration::from_secs(60),
169            epoch_max_bytes: DEFAULT_EPOCH_MAX_BYTES,
170            kafka_properties: HashMap::new(),
171        }
172    }
173}
174
175impl KafkaSinkConfig {
176    /// Parses a sink config from a resolved [`ConnectorConfig`].
177    ///
178    /// # Errors
179    ///
180    /// Returns `ConnectorError::MissingConfig` if required keys are absent,
181    /// or `ConnectorError::ConfigurationError` on invalid values.
182    #[allow(clippy::too_many_lines, clippy::field_reassign_with_default)]
183    pub fn from_config(config: &ConnectorConfig) -> Result<Self, ConnectorError> {
184        let mut cfg = Self::default();
185
186        cfg.bootstrap_servers = config
187            .get("bootstrap.servers")
188            .ok_or_else(|| ConnectorError::missing_config("bootstrap.servers"))?
189            .to_string();
190
191        cfg.topic = config
192            .get("topic")
193            .ok_or_else(|| ConnectorError::missing_config("topic"))?
194            .to_string();
195
196        if let Some(s) = config.get("security.protocol") {
197            cfg.security_protocol = s.parse()?;
198        }
199
200        if let Some(s) = config.get("sasl.mechanism") {
201            cfg.sasl_mechanism = Some(s.parse()?);
202        }
203
204        cfg.sasl_username = config.get("sasl.username").map(String::from);
205        cfg.sasl_password = config.get("sasl.password").map(String::from);
206        cfg.ssl_ca_location = config.get("ssl.ca.location").map(String::from);
207        cfg.ssl_certificate_location = config.get("ssl.certificate.location").map(String::from);
208        cfg.ssl_key_location = config.get("ssl.key.location").map(String::from);
209        cfg.ssl_key_password = config.get("ssl.key.password").map(String::from);
210
211        if let Some(fmt) = config.get("format") {
212            cfg.format = fmt.parse().map_err(ConnectorError::Serde)?;
213        }
214
215        cfg.schema_registry_url = config.get("schema.registry.url").map(String::from);
216
217        let sr_user = config.get("schema.registry.username");
218        let sr_pass = config.get("schema.registry.password");
219        if let (Some(user), Some(pass)) = (sr_user, sr_pass) {
220            cfg.schema_registry_auth = Some(SrAuth {
221                username: user.to_string(),
222                password: pass.to_string(),
223            });
224        }
225
226        if let Some(c) = config.get("schema.compatibility") {
227            cfg.schema_compatibility = Some(c.parse().map_err(|_| {
228                ConnectorError::ConfigurationError(format!("invalid schema.compatibility: '{c}'"))
229            })?);
230        }
231
232        cfg.schema_registry_ssl_ca_location = config
233            .get("schema.registry.ssl.ca.location")
234            .map(String::from);
235
236        if config.get("acks").is_some() {
237            return Err(ConnectorError::ConfigurationError(
238                "'acks' is not configurable; the durable Kafka sink always requires acks=all"
239                    .into(),
240            ));
241        }
242
243        if let Some(v) = config.get("max.in.flight.requests") {
244            cfg.max_in_flight = v.parse().map_err(|_| {
245                ConnectorError::ConfigurationError(format!("invalid max.in.flight.requests: '{v}'"))
246            })?;
247        }
248
249        if let Some(v) = config.get("delivery.timeout.ms") {
250            let ms: u64 = v.parse().map_err(|_| {
251                ConnectorError::ConfigurationError(format!("invalid delivery.timeout.ms: '{v}'"))
252            })?;
253            cfg.delivery_timeout = Duration::from_millis(ms);
254        }
255
256        cfg.key_column = config.get("key.column").map(String::from);
257        cfg.key_format = config
258            .get("key.format")
259            .map_or(Ok(KeyFormat::default()), str::parse)?;
260        cfg.partition_count = config.get_parsed::<i32>("partition.count")?;
261
262        cfg.envelope = match config
263            .get("envelope")
264            .map(str::to_ascii_lowercase)
265            .as_deref()
266        {
267            None | Some("append") => SinkEnvelope::Append,
268            Some("upsert") => SinkEnvelope::Upsert,
269            Some(other) => {
270                return Err(ConnectorError::ConfigurationError(format!(
271                    "unknown envelope '{other}' (expected 'append' or 'upsert')"
272                )));
273            }
274        };
275        if cfg.envelope == SinkEnvelope::Upsert
276            && cfg
277                .key_column
278                .as_deref()
279                .is_none_or(|k| k.trim().is_empty())
280        {
281            return Err(ConnectorError::ConfigurationError(
282                "envelope = 'upsert' requires a non-empty 'key.column' (the merge key)".into(),
283            ));
284        }
285
286        cfg.parse_batching(config)?;
287        cfg.parse_delivery(config)?;
288
289        cfg.dlq_topic = config.get("dlq.topic").map(String::from);
290        if cfg.dlq_topic.is_some() && cfg.envelope == SinkEnvelope::Upsert {
291            // Upsert failures poison the epoch; a lone tombstone in a DLQ would corrupt the
292            // compacted topic. Reject rather than silently ignore the DLQ.
293            return Err(ConnectorError::ConfigurationError(
294                "'dlq.topic' is not supported with envelope = 'upsert' (upsert failures poison the \
295                 epoch instead of routing to a DLQ)"
296                    .into(),
297            ));
298        }
299
300        for (key, value) in config.properties_with_prefix("kafka.") {
301            cfg.kafka_properties.insert(key, value);
302        }
303
304        cfg.validate()?;
305        Ok(cfg)
306    }
307
308    /// Producer batching: `latency.profile` defaults first, then explicit options override them.
309    fn parse_batching(&mut self, config: &ConnectorConfig) -> Result<(), ConnectorError> {
310        self.latency_profile = config
311            .get("latency.profile")
312            .map_or(Ok(LatencyProfile::default()), str::parse)?;
313        let profile = self.latency_profile.sink_defaults();
314        self.linger_ms = profile.linger_ms;
315        self.batch_size = profile.batch_size;
316        self.compression = profile.compression;
317
318        if let Some(p) = config.get("partitioner") {
319            self.partitioner = p.parse().map_err(|_| {
320                ConnectorError::ConfigurationError(format!(
321                    "invalid partitioner: '{p}' (expected 'key-hash', 'round-robin', or 'sticky')"
322                ))
323            })?;
324        }
325
326        if let Some(v) = config.get("linger.ms") {
327            self.linger_ms = v.parse().map_err(|_| {
328                ConnectorError::ConfigurationError(format!("invalid linger.ms: '{v}'"))
329            })?;
330        }
331
332        if let Some(v) = config.get("batch.size") {
333            self.batch_size = v.parse().map_err(|_| {
334                ConnectorError::ConfigurationError(format!("invalid batch.size: '{v}'"))
335            })?;
336        }
337
338        if let Some(v) = config.get("batch.num.messages") {
339            self.batch_num_messages = Some(v.parse().map_err(|_| {
340                ConnectorError::ConfigurationError(format!("invalid batch.num.messages: '{v}'"))
341            })?);
342        }
343
344        if let Some(c) = config.get("compression.type") {
345            self.compression = c.parse().map_err(|_| {
346                ConnectorError::ConfigurationError(format!("invalid compression.type: '{c}'"))
347            })?;
348        }
349        Ok(())
350    }
351
352    /// The injected delivery guarantee and the exactly-once publication options.
353    fn parse_delivery(&mut self, config: &ConnectorConfig) -> Result<(), ConnectorError> {
354        if let Some(delivery) = config.get("delivery.guarantee") {
355            self.delivery = delivery.parse().map_err(|error| {
356                ConnectorError::ConfigurationError(format!("invalid delivery.guarantee: {error}"))
357            })?;
358        }
359        self.commit_topic = config.get("commit.topic").map(String::from);
360        if let Some(ms) = config.get_parsed::<u64>("transaction.timeout.ms")? {
361            self.transaction_timeout = Duration::from_millis(ms);
362        }
363        if let Some(bytes) = config.get_parsed::<usize>("epoch.max.bytes")? {
364            self.epoch_max_bytes = bytes;
365        }
366        Ok(())
367    }
368
369    /// Validates the configuration.
370    ///
371    /// # Errors
372    ///
373    /// Returns `ConnectorError::ConfigurationError` on invalid combinations.
374    pub fn validate(&self) -> Result<(), ConnectorError> {
375        if self.bootstrap_servers.is_empty() {
376            return Err(ConnectorError::missing_config("bootstrap.servers"));
377        }
378        if self.topic.is_empty() {
379            return Err(ConnectorError::missing_config("topic"));
380        }
381
382        if self.security_protocol.uses_sasl() && self.sasl_mechanism.is_none() {
383            return Err(ConnectorError::ConfigurationError(
384                "sasl.mechanism is required when security.protocol is sasl_plaintext or sasl_ssl"
385                    .into(),
386            ));
387        }
388
389        if let Some(mechanism) = &self.sasl_mechanism {
390            if mechanism.requires_credentials()
391                && (self.sasl_username.is_none() || self.sasl_password.is_none())
392            {
393                return Err(ConnectorError::ConfigurationError(format!(
394                    "sasl.username and sasl.password are required for {mechanism} mechanism"
395                )));
396            }
397        }
398
399        if self.security_protocol.uses_ssl() {
400            if let Some(ref ca) = self.ssl_ca_location {
401                if ca.is_empty() {
402                    return Err(ConnectorError::ConfigurationError(
403                        "ssl.ca.location cannot be empty when specified".into(),
404                    ));
405                }
406            }
407        }
408
409        if self.format == Format::Debezium {
410            return Err(ConnectorError::ConfigurationError(
411                "Debezium is a deserialization-only format and cannot be used for sinks".into(),
412            ));
413        }
414        if self.format == Format::Fpml {
415            return Err(ConnectorError::ConfigurationError(
416                "FpML is a source-only format and cannot be used for sinks".into(),
417            ));
418        }
419
420        if self.partition_count.is_some_and(|count| count <= 0) {
421            return Err(ConnectorError::ConfigurationError(
422                "partition.count must be positive".into(),
423            ));
424        }
425        if self.envelope == SinkEnvelope::Upsert && self.partition_count.is_none() {
426            return Err(ConnectorError::ConfigurationError(
427                "envelope = 'upsert' requires partition.count: a compacted topic's keys must keep \
428                 their partition when partitions are added"
429                    .into(),
430            ));
431        }
432        if self.envelope == SinkEnvelope::Upsert && self.partitioner != PartitionStrategy::KeyHash {
433            return Err(ConnectorError::ConfigurationError(format!(
434                "envelope = 'upsert' requires partitioner = 'key-hash': '{}' ignores the key, so \
435                 a key's updates and tombstone could land in different partitions",
436                self.partitioner
437            )));
438        }
439
440        if self.format == Format::Avro && self.schema_registry_url.is_none() {
441            return Err(ConnectorError::ConfigurationError(
442                "Avro format requires 'schema.registry.url'".into(),
443            ));
444        }
445
446        if !(1..=5).contains(&self.max_in_flight) {
447            return Err(ConnectorError::ConfigurationError(
448                "max.in.flight.requests must be between 1 and 5 when idempotence is enabled".into(),
449            ));
450        }
451
452        if self.delivery_timeout.is_zero() {
453            return Err(ConnectorError::ConfigurationError(
454                "delivery.timeout.ms must be > 0; zero disables librdkafka's delivery deadline"
455                    .into(),
456            ));
457        }
458        if self.delivery_timeout > MAX_DELIVERY_TIMEOUT {
459            return Err(ConnectorError::ConfigurationError(format!(
460                "delivery.timeout.ms must be <= {}",
461                MAX_DELIVERY_TIMEOUT.as_millis()
462            )));
463        }
464
465        self.validate_exactly_once()?;
466        super::librdkafka::require_supported_client(
467            self.security_protocol,
468            self.sasl_mechanism,
469            Some(self.compression.as_rdkafka_str()),
470        )?;
471        super::librdkafka::validate_passthrough(
472            &self.owned_rdkafka_config(),
473            &self.kafka_properties,
474        )
475    }
476
477    fn validate_exactly_once(&self) -> Result<(), ConnectorError> {
478        if self.delivery != DeliveryGuarantee::ExactlyOnce {
479            if self.commit_topic.is_some() {
480                return Err(ConnectorError::ConfigurationError(
481                    "commit.topic is used only with exactly-once delivery".into(),
482                ));
483            }
484            return Ok(());
485        }
486        let commit_topic = self.commit_topic.as_deref().unwrap_or_default();
487        if commit_topic.is_empty() {
488            return Err(ConnectorError::ConfigurationError(
489                "exactly-once Kafka output requires 'commit.topic', a compacted topic that \
490                 records each committed checkpoint"
491                    .into(),
492            ));
493        }
494        if commit_topic == self.topic {
495            return Err(ConnectorError::ConfigurationError(
496                "commit.topic must differ from the output topic".into(),
497            ));
498        }
499        if self.dlq_topic.is_some() {
500            return Err(ConnectorError::ConfigurationError(
501                "'dlq.topic' is not supported with exactly-once delivery: dead-lettered records \
502                 would be visible outside the checkpoint transaction"
503                    .into(),
504            ));
505        }
506        if self.transaction_timeout.is_zero() || self.transaction_timeout > MAX_TRANSACTION_TIMEOUT
507        {
508            return Err(ConnectorError::ConfigurationError(format!(
509                "transaction.timeout.ms must be in 1..={}",
510                MAX_TRANSACTION_TIMEOUT.as_millis()
511            )));
512        }
513        if self.epoch_max_bytes == 0 || self.epoch_max_bytes > MAX_EPOCH_BYTES {
514            return Err(ConnectorError::ConfigurationError(format!(
515                "epoch.max.bytes must be in 1..={MAX_EPOCH_BYTES}"
516            )));
517        }
518        self.message_max_bytes()?;
519        Ok(())
520    }
521
522    /// The producer's largest record, from `kafka.message.max.bytes` or librdkafka's default.
523    ///
524    /// # Errors
525    /// Returns a configuration error when the passthrough value is not a positive integer.
526    pub fn message_max_bytes(&self) -> Result<usize, ConnectorError> {
527        const LIBRDKAFKA_DEFAULT: usize = 1_000_000;
528        self.kafka_properties
529            .get("message.max.bytes")
530            .map_or(Ok(LIBRDKAFKA_DEFAULT), |value| {
531                value
532                    .parse::<usize>()
533                    .ok()
534                    .filter(|bytes| *bytes > 0)
535                    .ok_or_else(|| {
536                        ConnectorError::ConfigurationError(format!(
537                            "invalid kafka.message.max.bytes: '{value}'"
538                        ))
539                    })
540            })
541    }
542
543    /// Broker address and transport security, shared by every client the sink creates.
544    #[must_use]
545    pub fn connection_rdkafka_config(&self) -> ClientConfig {
546        let mut config = ClientConfig::new();
547        config.set("bootstrap.servers", &self.bootstrap_servers);
548        config.set("security.protocol", self.security_protocol.as_rdkafka_str());
549        for (key, value) in [
550            (
551                "sasl.mechanism",
552                self.sasl_mechanism
553                    .map(|mechanism| mechanism.as_rdkafka_str()),
554            ),
555            ("sasl.username", self.sasl_username.as_deref()),
556            ("sasl.password", self.sasl_password.as_deref()),
557            ("ssl.ca.location", self.ssl_ca_location.as_deref()),
558            (
559                "ssl.certificate.location",
560                self.ssl_certificate_location.as_deref(),
561            ),
562            ("ssl.key.location", self.ssl_key_location.as_deref()),
563            ("ssl.key.password", self.ssl_key_password.as_deref()),
564        ] {
565            if let Some(value) = value {
566                config.set(key, value);
567            }
568        }
569        config
570    }
571
572    /// The exactly-once committer's producer: the configured producer settings plus a stable
573    /// `transactional.id`, with the delivery deadline capped by the transaction timeout as
574    /// librdkafka requires.
575    #[must_use]
576    pub fn transactional_rdkafka_config(&self, transactional_id: &str) -> ClientConfig {
577        let mut config = self.to_rdkafka_config();
578        let message_timeout = self.delivery_timeout.min(self.transaction_timeout);
579        config
580            .set("transactional.id", transactional_id)
581            .set(
582                "transaction.timeout.ms",
583                self.transaction_timeout.as_millis().to_string(),
584            )
585            .set(
586                "message.timeout.ms",
587                message_timeout.as_millis().to_string(),
588            );
589        config
590    }
591
592    /// Configured key column names, in key order.
593    #[must_use]
594    pub fn key_columns(&self) -> Vec<String> {
595        self.key_column
596            .as_deref()
597            .map(|columns| {
598                columns
599                    .split(',')
600                    .map(str::trim)
601                    .filter(|column| !column.is_empty())
602                    .map(String::from)
603                    .collect()
604            })
605            .unwrap_or_default()
606    }
607
608    /// Builds the rdkafka [`ClientConfig`]: connector-owned settings, then the validated
609    /// `kafka.*` passthrough properties.
610    ///
611    /// Always sets `enable.idempotence=true` for durable at-least-once writes.
612    #[must_use]
613    pub fn to_rdkafka_config(&self) -> ClientConfig {
614        let mut config = self.owned_rdkafka_config();
615        super::librdkafka::apply_passthrough(&mut config, &self.kafka_properties);
616        config
617    }
618
619    fn owned_rdkafka_config(&self) -> ClientConfig {
620        let mut config = self.connection_rdkafka_config();
621        config
622            .set("enable.idempotence", "true")
623            .set("acks", "all")
624            .set("linger.ms", self.linger_ms.to_string())
625            .set("batch.size", self.batch_size.to_string())
626            .set("compression.type", self.compression.as_rdkafka_str())
627            .set(
628                "max.in.flight.requests.per.connection",
629                self.max_in_flight.to_string(),
630            )
631            .set(
632                "message.timeout.ms",
633                self.delivery_timeout.as_millis().to_string(),
634            );
635
636        if let Some(num_msgs) = self.batch_num_messages {
637            config.set("batch.num.messages", num_msgs.to_string());
638        }
639
640        config
641    }
642
643    /// Builds an rdkafka [`ClientConfig`] for the dead letter queue producer.
644    ///
645    /// Inherits security settings (SASL, SSL) from the main config but is
646    /// non-transactional. Does not set `transactional.id`.
647    #[must_use]
648    pub fn to_dlq_rdkafka_config(&self) -> ClientConfig {
649        let mut config = self.connection_rdkafka_config();
650        config
651            .set("enable.idempotence", "true")
652            .set("acks", "all")
653            .set(
654                "max.in.flight.requests.per.connection",
655                self.max_in_flight.to_string(),
656            )
657            .set(
658                "message.timeout.ms",
659                self.delivery_timeout.as_millis().to_string(),
660            );
661
662        config
663    }
664}
665
666/// How `key.column` values become Kafka record-key bytes.
667#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
668pub enum KeyFormat {
669    /// UTF-8 for string columns and raw bytes for binary columns, as Kafka's Java
670    /// `StringSerializer` and `ByteArraySerializer` write them. Other types need an explicit format.
671    #[default]
672    Native,
673    /// The SQL text rendering of one column of any type.
674    Text,
675    /// Kafka's Java serializer bytes: `Native` plus big-endian `Short`, `Integer`, `Long`, `Float`
676    /// and `Double`.
677    Kafka,
678    /// A JSON object of one or more key columns with NULLs written explicitly; never a NULL key.
679    Json,
680}
681
682str_enum!(KeyFormat, lowercase, ConnectorError, "invalid key.format",
683    Native => "native";
684    Text => "text";
685    Kafka => "kafka";
686    Json => "json"
687);
688
689impl KeyFormat {
690    /// Whether this format can encode a key column of `data_type`.
691    #[must_use]
692    pub fn encodes(self, data_type: &DataType) -> bool {
693        let native = matches!(
694            data_type,
695            DataType::Utf8
696                | DataType::LargeUtf8
697                | DataType::Utf8View
698                | DataType::Binary
699                | DataType::LargeBinary
700                | DataType::BinaryView
701                | DataType::FixedSizeBinary(_)
702        );
703        match self {
704            Self::Native => native,
705            Self::Kafka => {
706                native
707                    || matches!(
708                        data_type,
709                        DataType::Int16
710                            | DataType::Int32
711                            | DataType::Int64
712                            | DataType::Float32
713                            | DataType::Float64
714                    )
715            }
716            Self::Text | Self::Json => true,
717        }
718    }
719
720    pub(crate) const fn remedy(self) -> &'static str {
721        match self {
722            Self::Native => {
723                "set key.format to 'kafka' (Java serializer bytes), 'text' (SQL text) or 'json'"
724            }
725            Self::Kafka => "use key.format = 'text' or 'json' for this type",
726            Self::Text | Self::Json => "choose a supported key column type",
727        }
728    }
729}
730
731/// Partitioning strategy for distributing records across Kafka partitions.
732#[derive(Debug, Clone, Copy, PartialEq, Eq)]
733pub enum PartitionStrategy {
734    /// Hash the key column (Murmur2, Kafka-compatible).
735    KeyHash,
736    /// Round-robin across all partitions.
737    RoundRobin,
738    /// Sticky: batch records to the same partition until full.
739    Sticky,
740}
741
742str_enum!(PartitionStrategy, lowercase_udash, String, "unknown partition strategy",
743    KeyHash => "key-hash", "keyhash", "hash";
744    RoundRobin => "round-robin", "roundrobin";
745    Sticky => "sticky"
746);
747
748/// Compression type for produced Kafka messages.
749#[derive(Debug, Clone, Copy, PartialEq, Eq)]
750pub enum CompressionType {
751    /// No compression.
752    None,
753    /// Gzip compression.
754    Gzip,
755    /// Snappy compression.
756    Snappy,
757    /// LZ4 compression.
758    Lz4,
759    /// Zstandard compression.
760    Zstd,
761}
762
763impl CompressionType {
764    /// Returns the rdkafka configuration string.
765    #[must_use]
766    pub fn as_rdkafka_str(&self) -> &'static str {
767        match self {
768            Self::None => "none",
769            Self::Gzip => "gzip",
770            Self::Snappy => "snappy",
771            Self::Lz4 => "lz4",
772            Self::Zstd => "zstd",
773        }
774    }
775}
776
777str_enum!(fromstr CompressionType, lowercase_nodash, String, "unknown compression type",
778    None => "none";
779    Gzip => "gzip";
780    Snappy => "snappy";
781    Lz4 => "lz4";
782    Zstd => "zstd", "zstandard"
783);
784
785impl std::fmt::Display for CompressionType {
786    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
787        write!(f, "{}", self.as_rdkafka_str())
788    }
789}
790
791#[cfg(test)]
792mod tests;
laminar_connectors/kafka/sink_config/mod.rs

laminar_connectors/kafka/sink_config/
mod.rs
