laminar_connectors/kafka/config/
mod.rs

1//! Kafka source config — broker connection, format, Schema Registry,
2//! backpressure, and pass-through `rdkafka` properties.
3
4use std::collections::HashMap;
5use std::time::Duration;
6
7use rdkafka::config::ClientConfig;
8
9use crate::error::ConnectorError;
10use crate::serde::Format;
11
12mod source_options;
13
14/// Kafka security protocol for broker connections.
15///
16/// Determines encryption (SSL/TLS) and authentication (SASL) requirements.
17#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
18pub enum SecurityProtocol {
19    /// Plain-text communication (no encryption, no authentication).
20    #[default]
21    Plaintext,
22    /// SSL/TLS encryption without SASL authentication.
23    Ssl,
24    /// SASL authentication over plain-text connection.
25    SaslPlaintext,
26    /// SASL authentication over SSL/TLS encrypted connection.
27    SaslSsl,
28}
29
30impl SecurityProtocol {
31    /// Returns the rdkafka config value string.
32    #[must_use]
33    pub fn as_rdkafka_str(&self) -> &'static str {
34        match self {
35            SecurityProtocol::Plaintext => "plaintext",
36            SecurityProtocol::Ssl => "ssl",
37            SecurityProtocol::SaslPlaintext => "sasl_plaintext",
38            SecurityProtocol::SaslSsl => "sasl_ssl",
39        }
40    }
41
42    /// Returns true if this protocol uses SSL/TLS.
43    #[must_use]
44    pub fn uses_ssl(&self) -> bool {
45        matches!(self, SecurityProtocol::Ssl | SecurityProtocol::SaslSsl)
46    }
47
48    /// Returns true if this protocol uses SASL authentication.
49    #[must_use]
50    pub fn uses_sasl(&self) -> bool {
51        matches!(
52            self,
53            SecurityProtocol::SaslPlaintext | SecurityProtocol::SaslSsl
54        )
55    }
56}
57
58str_enum!(fromstr SecurityProtocol, lowercase, ConnectorError, "invalid security.protocol",
59    Plaintext => "plaintext";
60    Ssl => "ssl";
61    SaslPlaintext => "sasl_plaintext";
62    SaslSsl => "sasl_ssl"
63);
64
65impl std::fmt::Display for SecurityProtocol {
66    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
67        write!(f, "{}", self.as_rdkafka_str())
68    }
69}
70
71/// SASL authentication mechanism for Kafka.
72#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
73pub enum SaslMechanism {
74    /// PLAIN: Simple username/password authentication.
75    #[default]
76    Plain,
77    /// SCRAM-SHA-256: Salted Challenge Response Authentication Mechanism.
78    ScramSha256,
79    /// SCRAM-SHA-512: Salted Challenge Response Authentication Mechanism (stronger).
80    ScramSha512,
81    /// GSSAPI: Kerberos authentication.
82    Gssapi,
83    /// OAUTHBEARER: OAuth 2.0 bearer token authentication.
84    Oauthbearer,
85}
86
87impl SaslMechanism {
88    /// Returns the rdkafka config value string.
89    #[must_use]
90    pub fn as_rdkafka_str(&self) -> &'static str {
91        match self {
92            SaslMechanism::Plain => "PLAIN",
93            SaslMechanism::ScramSha256 => "SCRAM-SHA-256",
94            SaslMechanism::ScramSha512 => "SCRAM-SHA-512",
95            SaslMechanism::Gssapi => "GSSAPI",
96            SaslMechanism::Oauthbearer => "OAUTHBEARER",
97        }
98    }
99
100    /// Returns true if this mechanism requires username/password.
101    #[must_use]
102    pub fn requires_credentials(&self) -> bool {
103        matches!(
104            self,
105            SaslMechanism::Plain | SaslMechanism::ScramSha256 | SaslMechanism::ScramSha512
106        )
107    }
108}
109
110str_enum!(fromstr SaslMechanism, uppercase, ConnectorError, "invalid sasl.mechanism",
111    Plain => "PLAIN";
112    ScramSha256 => "SCRAM_SHA_256", "SCRAM_SHA256";
113    ScramSha512 => "SCRAM_SHA_512", "SCRAM_SHA512";
114    Gssapi => "GSSAPI", "KERBEROS";
115    Oauthbearer => "OAUTHBEARER", "OAUTH"
116);
117
118impl std::fmt::Display for SaslMechanism {
119    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
120        write!(f, "{}", self.as_rdkafka_str())
121    }
122}
123
124/// Consumer isolation level for reading transactional messages.
125///
126/// Controls whether to read uncommitted messages from transactional producers.
127#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
128pub enum IsolationLevel {
129    /// Read all messages including uncommitted transactional messages.
130    ReadUncommitted,
131    /// Only read committed messages (recommended for transactional pipelines).
132    #[default]
133    ReadCommitted,
134}
135
136impl IsolationLevel {
137    /// Returns the rdkafka config value string.
138    #[must_use]
139    pub fn as_rdkafka_str(&self) -> &'static str {
140        match self {
141            IsolationLevel::ReadUncommitted => "read_uncommitted",
142            IsolationLevel::ReadCommitted => "read_committed",
143        }
144    }
145}
146
147str_enum!(fromstr IsolationLevel, lowercase, ConnectorError, "invalid isolation.level",
148    ReadUncommitted => "read_uncommitted";
149    ReadCommitted => "read_committed"
150);
151
152impl std::fmt::Display for IsolationLevel {
153    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
154        write!(f, "{}", self.as_rdkafka_str())
155    }
156}
157
158/// Consumer startup mode controlling where to begin consuming.
159///
160/// This is a higher-level abstraction than `auto.offset.reset` that provides
161/// more control over initial positioning, including timestamp-based and
162/// partition-specific offset assignment.
163#[derive(Debug, Clone, PartialEq, Eq, Default)]
164pub enum StartupMode {
165    /// Use committed group offsets, fall back to `auto.offset.reset` if none exist.
166    #[default]
167    GroupOffsets,
168    /// Start from the earliest available offset in each partition.
169    Earliest,
170    /// Start from the latest offset in each partition (only new messages).
171    Latest,
172    /// Start from specific offsets per partition (`partition_id` -> offset).
173    SpecificOffsets(HashMap<i32, i64>),
174    /// Start from a specific timestamp (milliseconds since epoch).
175    /// The consumer seeks to the first message with timestamp >= this value.
176    Timestamp(i64),
177}
178
179impl std::str::FromStr for StartupMode {
180    type Err = ConnectorError;
181
182    fn from_str(s: &str) -> Result<Self, Self::Err> {
183        match s.to_lowercase().replace('-', "_").as_str() {
184            "group_offsets" | "group" => Ok(StartupMode::GroupOffsets),
185            "earliest" => Ok(StartupMode::Earliest),
186            "latest" => Ok(StartupMode::Latest),
187            "timestamp" => Err(ConnectorError::ConfigurationError(
188                "use 'startup.timestamp.ms' to start from a timestamp, \
189                 not 'startup.mode = timestamp'"
190                    .into(),
191            )),
192            other => Err(ConnectorError::ConfigurationError(format!(
193                "invalid startup.mode: '{other}' (expected group-offsets/earliest/latest)"
194            ))),
195        }
196    }
197}
198
199impl std::fmt::Display for StartupMode {
200    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
201        match self {
202            StartupMode::GroupOffsets => write!(f, "group-offsets"),
203            StartupMode::Earliest => write!(f, "earliest"),
204            StartupMode::Latest => write!(f, "latest"),
205            StartupMode::SpecificOffsets(offsets) => {
206                write!(f, "specific-offsets({} partitions)", offsets.len())
207            }
208            StartupMode::Timestamp(ts) => write!(f, "timestamp({ts})"),
209        }
210    }
211}
212
213/// Topic subscription mode: explicit list or regex pattern.
214#[derive(Debug, Clone, PartialEq, Eq)]
215pub enum TopicSubscription {
216    /// Subscribe to a specific list of topic names.
217    Topics(Vec<String>),
218    /// Subscribe to topics matching a regex pattern (e.g., `events-.*`).
219    Pattern(String),
220}
221
222impl TopicSubscription {
223    /// Returns the topic names if this is a `Topics` subscription.
224    #[must_use]
225    pub fn topics(&self) -> Option<&[String]> {
226        match self {
227            TopicSubscription::Topics(t) => Some(t),
228            TopicSubscription::Pattern(_) => None,
229        }
230    }
231
232    /// Returns the pattern if this is a `Pattern` subscription.
233    #[must_use]
234    pub fn pattern(&self) -> Option<&str> {
235        match self {
236            TopicSubscription::Topics(_) => None,
237            TopicSubscription::Pattern(p) => Some(p),
238        }
239    }
240
241    /// Returns true if this is a pattern-based subscription.
242    #[must_use]
243    pub fn is_pattern(&self) -> bool {
244        matches!(self, TopicSubscription::Pattern(_))
245    }
246}
247
248impl Default for TopicSubscription {
249    fn default() -> Self {
250        TopicSubscription::Topics(Vec::new())
251    }
252}
253
254/// Auto-offset reset policy for new consumer groups.
255#[derive(Debug, Clone, Copy, PartialEq, Eq)]
256pub enum OffsetReset {
257    /// Start from the earliest available offset.
258    Earliest,
259    /// Start from the latest offset (only new messages).
260    Latest,
261    /// Fail if no committed offset exists.
262    None,
263}
264
265impl OffsetReset {
266    /// Returns the rdkafka config value string.
267    #[must_use]
268    pub fn as_rdkafka_str(&self) -> &'static str {
269        match self {
270            OffsetReset::Earliest => "earliest",
271            OffsetReset::Latest => "latest",
272            OffsetReset::None => "error",
273        }
274    }
275}
276
277str_enum!(fromstr OffsetReset, lowercase_nodash, ConnectorError, "invalid auto.offset.reset",
278    Earliest => "earliest", "beginning";
279    Latest => "latest", "end";
280    None => "none", "error"
281);
282
283/// Kafka partition assignment strategy.
284#[derive(Debug, Clone, Copy, PartialEq, Eq)]
285pub enum AssignmentStrategy {
286    /// Range assignment (default).
287    Range,
288    /// Round-robin assignment.
289    RoundRobin,
290    /// Cooperative sticky assignment.
291    CooperativeSticky,
292}
293
294impl AssignmentStrategy {
295    /// Returns the rdkafka config value string.
296    #[must_use]
297    pub fn as_rdkafka_str(&self) -> &'static str {
298        match self {
299            AssignmentStrategy::Range => "range",
300            AssignmentStrategy::RoundRobin => "roundrobin",
301            AssignmentStrategy::CooperativeSticky => "cooperative-sticky",
302        }
303    }
304}
305
306str_enum!(fromstr AssignmentStrategy, lowercase_nodash, ConnectorError,
307    "invalid partition.assignment.strategy",
308    Range => "range";
309    RoundRobin => "roundrobin", "round-robin", "round_robin";
310    CooperativeSticky => "cooperative-sticky", "cooperative_sticky"
311);
312
313/// Schema Registry compatibility level.
314#[derive(Debug, Clone, Copy, PartialEq, Eq)]
315pub enum CompatibilityLevel {
316    /// New schema can read old data.
317    Backward,
318    /// Backward compatible with all prior versions.
319    BackwardTransitive,
320    /// Old schema can read new data.
321    Forward,
322    /// Forward compatible with all prior versions.
323    ForwardTransitive,
324    /// Both backward and forward compatible.
325    Full,
326    /// Full compatible with all prior versions.
327    FullTransitive,
328    /// No compatibility checking.
329    None,
330}
331
332impl CompatibilityLevel {
333    /// Returns the Schema Registry API string.
334    #[must_use]
335    pub fn as_str(&self) -> &'static str {
336        match self {
337            CompatibilityLevel::Backward => "BACKWARD",
338            CompatibilityLevel::BackwardTransitive => "BACKWARD_TRANSITIVE",
339            CompatibilityLevel::Forward => "FORWARD",
340            CompatibilityLevel::ForwardTransitive => "FORWARD_TRANSITIVE",
341            CompatibilityLevel::Full => "FULL",
342            CompatibilityLevel::FullTransitive => "FULL_TRANSITIVE",
343            CompatibilityLevel::None => "NONE",
344        }
345    }
346}
347
348str_enum!(fromstr CompatibilityLevel, uppercase, ConnectorError, "invalid schema.compatibility",
349    Backward => "BACKWARD";
350    BackwardTransitive => "BACKWARD_TRANSITIVE";
351    Forward => "FORWARD";
352    ForwardTransitive => "FORWARD_TRANSITIVE";
353    Full => "FULL";
354    FullTransitive => "FULL_TRANSITIVE";
355    None => "NONE"
356);
357
358impl std::fmt::Display for CompatibilityLevel {
359    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
360        write!(f, "{}", self.as_str())
361    }
362}
363
364/// Strategy for handling Avro schema evolution at runtime.
365#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
366pub enum SchemaEvolutionStrategy {
367    /// Log schema changes, continue processing.
368    #[default]
369    Log,
370    /// Return an error on incompatible schema changes.
371    Reject,
372    /// No detection — skip schema diffing entirely.
373    Ignore,
374}
375
376str_enum!(fromstr SchemaEvolutionStrategy, lowercase, ConnectorError,
377    "invalid schema.evolution.strategy",
378    Log => "log";
379    Reject => "reject";
380    Ignore => "ignore"
381);
382
383impl std::fmt::Display for SchemaEvolutionStrategy {
384    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
385        match self {
386            SchemaEvolutionStrategy::Log => write!(f, "log"),
387            SchemaEvolutionStrategy::Reject => write!(f, "reject"),
388            SchemaEvolutionStrategy::Ignore => write!(f, "ignore"),
389        }
390    }
391}
392
393/// Confluent subject-name strategy.
394#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
395pub enum SubjectNameStrategy {
396    /// `{topic}-value` (Confluent default).
397    #[default]
398    TopicName,
399    /// `{record_name}-value`. Requires `schema.registry.record.name`.
400    RecordName,
401    /// `{topic}-{record_name}-value`. Requires `schema.registry.record.name`.
402    TopicRecordName,
403}
404
405str_enum!(fromstr SubjectNameStrategy, lowercase_nodash, ConnectorError,
406    "invalid schema.registry.subject.name.strategy",
407    TopicName => "topic-name", "topicname", "topicnamestrategy";
408    RecordName => "record-name", "recordname", "recordnamestrategy";
409    TopicRecordName => "topic-record-name", "topicrecordname", "topicrecordnamestrategy"
410);
411
412impl std::fmt::Display for SubjectNameStrategy {
413    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
414        match self {
415            SubjectNameStrategy::TopicName => write!(f, "topic-name"),
416            SubjectNameStrategy::RecordName => write!(f, "record-name"),
417            SubjectNameStrategy::TopicRecordName => write!(f, "topic-record-name"),
418        }
419    }
420}
421
422/// Build the SR `-value` subject for a topic. `from_config` validates
423/// that `record_name` is present for the record-based strategies, so
424/// the `expect`s are unreachable in practice.
425pub(crate) fn resolve_value_subject(
426    strategy: SubjectNameStrategy,
427    record_name: Option<&str>,
428    topic: &str,
429) -> String {
430    let name = || record_name.expect("from_config validates record.name");
431    match strategy {
432        SubjectNameStrategy::TopicName => format!("{topic}-value"),
433        SubjectNameStrategy::RecordName => format!("{}-value", name()),
434        SubjectNameStrategy::TopicRecordName => format!("{topic}-{}-value", name()),
435    }
436}
437
438/// Maps the Kafka-level [`CompatibilityLevel`] to the schema module's
439/// `CompatibilityMode` for evolution evaluation.
440impl From<CompatibilityLevel> for crate::schema::traits::CompatibilityMode {
441    fn from(level: CompatibilityLevel) -> Self {
442        use crate::schema::traits::CompatibilityMode;
443        match level {
444            CompatibilityLevel::Backward => CompatibilityMode::Backward,
445            CompatibilityLevel::BackwardTransitive => CompatibilityMode::BackwardTransitive,
446            CompatibilityLevel::Forward => CompatibilityMode::Forward,
447            CompatibilityLevel::ForwardTransitive => CompatibilityMode::ForwardTransitive,
448            CompatibilityLevel::Full => CompatibilityMode::Full,
449            CompatibilityLevel::FullTransitive => CompatibilityMode::FullTransitive,
450            CompatibilityLevel::None => CompatibilityMode::None,
451        }
452    }
453}
454
455/// Schema Registry authentication credentials.
456#[derive(Clone)]
457pub struct SrAuth {
458    /// Basic auth username.
459    pub username: String,
460    /// Basic auth password.
461    pub password: String,
462}
463
464impl std::fmt::Debug for SrAuth {
465    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
466        f.debug_struct("SrAuth")
467            .field("username", &self.username)
468            .field("password", &"***")
469            .finish()
470    }
471}
472
473/// Kafka source connector configuration.
474///
475/// Uses a custom `Debug` impl that redacts `sasl_password` and
476/// `ssl_key_password` to prevent credential leakage in logs.
477#[derive(Clone)]
478#[allow(clippy::struct_excessive_bools)] // Config struct — each bool is an independent user-facing knob.
479pub struct KafkaSourceConfig {
480    // -- Required --
481    /// Comma-separated list of broker addresses.
482    pub bootstrap_servers: String,
483    /// Consumer group identifier.
484    pub group_id: String,
485    /// Topic subscription (explicit list or regex pattern).
486    pub subscription: TopicSubscription,
487
488    // -- Security --
489    /// Security protocol for broker connections.
490    pub security_protocol: SecurityProtocol,
491    /// SASL authentication mechanism.
492    pub sasl_mechanism: Option<SaslMechanism>,
493    /// SASL username (for PLAIN, SCRAM-SHA-256, SCRAM-SHA-512).
494    pub sasl_username: Option<String>,
495    /// SASL password (for PLAIN, SCRAM-SHA-256, SCRAM-SHA-512).
496    pub sasl_password: Option<String>,
497    /// Path to SSL CA certificate file (PEM format).
498    pub ssl_ca_location: Option<String>,
499    /// Path to client SSL certificate file (PEM format).
500    pub ssl_certificate_location: Option<String>,
501    /// Path to client SSL private key file (PEM format).
502    pub ssl_key_location: Option<String>,
503    /// Password for encrypted SSL private key.
504    pub ssl_key_password: Option<String>,
505
506    // -- Format & Schema --
507    /// Data format for deserialization.
508    pub format: Format,
509    /// Confluent Schema Registry URL.
510    pub schema_registry_url: Option<String>,
511    /// Schema Registry authentication credentials.
512    pub schema_registry_auth: Option<SrAuth>,
513    /// Override compatibility level for the subject.
514    pub schema_compatibility: Option<CompatibilityLevel>,
515    /// How to handle Avro schema evolution detected at runtime.
516    pub schema_evolution_strategy: SchemaEvolutionStrategy,
517    /// Schema Registry SSL CA certificate path.
518    pub schema_registry_ssl_ca_location: Option<String>,
519    /// Schema Registry SSL client certificate path.
520    pub schema_registry_ssl_certificate_location: Option<String>,
521    /// Schema Registry SSL client key path.
522    pub schema_registry_ssl_key_location: Option<String>,
523    /// Confluent subject-name strategy. Default: `TopicName`.
524    pub schema_registry_subject_strategy: SubjectNameStrategy,
525    /// Record name for `record-name` / `topic-record-name` strategies.
526    pub schema_registry_record_name: Option<String>,
527    /// Deadline for Schema Registry lookups performed during schema
528    /// auto-discovery at DDL time. Default: 10s.
529    pub schema_registry_discovery_timeout: Duration,
530    /// Whether to include Kafka metadata columns (_partition, _offset, _timestamp).
531    pub include_metadata: bool,
532    /// Whether to include Kafka headers as a map column (_headers).
533    pub include_headers: bool,
534
535    // -- Consumer tuning --
536    /// Consumer startup mode (controls initial offset positioning).
537    pub startup_mode: StartupMode,
538    /// Where to start reading when no committed offset exists.
539    pub auto_offset_reset: OffsetReset,
540    /// Consumer transaction isolation level.
541    pub isolation_level: IsolationLevel,
542    /// Maximum records per poll batch.
543    pub max_poll_records: usize,
544    /// Partition assignment strategy.
545    pub partition_assignment_strategy: AssignmentStrategy,
546    /// Minimum bytes to return from a fetch (allows batching).
547    pub fetch_min_bytes: Option<i32>,
548    /// Maximum bytes to return from broker per request.
549    pub fetch_max_bytes: Option<i32>,
550    /// Maximum time broker waits for fetch.min.bytes.
551    pub fetch_max_wait_ms: Option<i32>,
552    /// Maximum bytes per partition to return from broker.
553    pub max_partition_fetch_bytes: Option<i32>,
554
555    // -- Consumer group timing --
556    /// Consumer session timeout (default: 45s — production-safe; rdkafka's
557    /// aggressive 10s default causes rebalance storms under GC pauses).
558    pub session_timeout: Duration,
559    /// Consumer heartbeat interval (default: 10s). Must satisfy
560    /// `heartbeat_interval * 3 < session_timeout` per Kafka broker requirement.
561    pub heartbeat_interval: Duration,
562    /// Maximum interval between calls to `rd_kafka_consumer_poll()` before
563    /// the broker considers this consumer dead and triggers a rebalance.
564    ///
565    /// Default: 600s (10 minutes). rdkafka's default is 300s, but with
566    /// reader-side pause/resume the reader keeps polling even under
567    /// backpressure, so this is a safety margin. Must be >= 60s.
568    pub max_poll_interval: Duration,
569    /// Maximum per-partition pre-fetch queue size in kbytes (default: 16384 = 16MB).
570    /// rdkafka's 64MB default is too aggressive for an embedded database.
571    pub queued_max_messages_kbytes: u32,
572
573    // -- Broker commit --
574    /// Whether to commit consumed offsets to the Kafka broker after each
575    /// `LaminarDB` checkpoint completes. Default: `true`.
576    ///
577    /// Advisory — `LaminarDB` recovery uses its own manifest, not broker-stored
578    /// offsets. This exists only so external tooling (`kafka-consumer-groups`,
579    /// kafka-exporter, Burrow) can observe progress. Guaranteed delivery always
580    /// derives an unrecorded partition's start from engine configuration and
581    /// deliberately ignores broker-stored group offsets from abandoned timelines.
582    pub broker_commit_on_checkpoint: bool,
583
584    // -- Backpressure --
585    /// Capacity of the bounded channel between the background Kafka reader
586    /// task and `poll_batch()` (default: 8192). Must be >= `max_poll_records`.
587    pub reader_channel_capacity: usize,
588    /// Channel fill ratio at which to pause consumption.
589    pub backpressure_high_watermark: f64,
590    /// Channel fill ratio at which to resume consumption.
591    pub backpressure_low_watermark: f64,
592
593    // -- Error handling --
594    /// Maximum tolerated deserialization error rate per `BestEffort` batch (0.0-1.0).
595    ///
596    /// When the poison pill fallback is active and the error rate exceeds
597    /// this threshold, the batch is rejected instead of returning partial
598    /// results. Guaranteed-delivery modes reject any deserialization failure;
599    /// they never use this threshold. Default: 0.5 (50%).
600    pub max_deser_error_rate: f64,
601
602    // -- Pass-through --
603    /// Additional rdkafka properties passed directly to librdkafka.
604    pub kafka_properties: HashMap<String, String>,
605}
606
607impl std::fmt::Debug for KafkaSourceConfig {
608    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
609        f.debug_struct("KafkaSourceConfig")
610            .field("bootstrap_servers", &self.bootstrap_servers)
611            .field("group_id", &self.group_id)
612            .field("subscription", &self.subscription)
613            .field("format", &self.format)
614            .field("security_protocol", &self.security_protocol)
615            .field("sasl_mechanism", &self.sasl_mechanism)
616            .field("sasl_username", &self.sasl_username)
617            .field("sasl_password", &self.sasl_password.as_ref().map(|_| "***"))
618            .field(
619                "ssl_key_password",
620                &self.ssl_key_password.as_ref().map(|_| "***"),
621            )
622            .field("max_poll_records", &self.max_poll_records)
623            .finish_non_exhaustive()
624    }
625}
626
627impl Default for KafkaSourceConfig {
628    fn default() -> Self {
629        Self {
630            bootstrap_servers: String::new(),
631            group_id: String::new(),
632            subscription: TopicSubscription::default(),
633            security_protocol: SecurityProtocol::default(),
634            sasl_mechanism: None,
635            sasl_username: None,
636            sasl_password: None,
637            ssl_ca_location: None,
638            ssl_certificate_location: None,
639            ssl_key_location: None,
640            ssl_key_password: None,
641            format: Format::Json,
642            schema_registry_url: None,
643            schema_registry_auth: None,
644            schema_compatibility: None,
645            schema_evolution_strategy: SchemaEvolutionStrategy::default(),
646            schema_registry_ssl_ca_location: None,
647            schema_registry_ssl_certificate_location: None,
648            schema_registry_ssl_key_location: None,
649            schema_registry_subject_strategy: SubjectNameStrategy::default(),
650            schema_registry_record_name: None,
651            schema_registry_discovery_timeout: Duration::from_secs(10),
652            include_metadata: false,
653            include_headers: false,
654            startup_mode: StartupMode::default(),
655            auto_offset_reset: OffsetReset::Earliest,
656            isolation_level: IsolationLevel::default(),
657            max_poll_records: 1000,
658            partition_assignment_strategy: AssignmentStrategy::Range,
659            fetch_min_bytes: None,
660            fetch_max_bytes: None,
661            fetch_max_wait_ms: None,
662            max_partition_fetch_bytes: None,
663            session_timeout: Duration::from_secs(45),
664            heartbeat_interval: Duration::from_secs(10),
665            max_poll_interval: Duration::from_secs(600),
666            queued_max_messages_kbytes: 16384,
667            broker_commit_on_checkpoint: true,
668            reader_channel_capacity: 8192,
669            backpressure_high_watermark: 0.8,
670            backpressure_low_watermark: 0.25,
671            max_deser_error_rate: 0.5,
672            kafka_properties: HashMap::new(),
673        }
674    }
675}
676
677impl KafkaSourceConfig {
678    /// Validates the configuration.
679    ///
680    /// # Errors
681    ///
682    /// Returns `ConnectorError::ConfigurationError` if the configuration is invalid.
683    pub fn validate(&self) -> Result<(), ConnectorError> {
684        if self.bootstrap_servers.is_empty() {
685            return Err(ConnectorError::ConfigurationError(
686                "bootstrap.servers cannot be empty".into(),
687            ));
688        }
689        if self.group_id.is_empty() {
690            return Err(ConnectorError::ConfigurationError(
691                "group.id cannot be empty".into(),
692            ));
693        }
694
695        // Validate topic subscription
696        match &self.subscription {
697            TopicSubscription::Topics(t) if t.is_empty() => {
698                return Err(ConnectorError::ConfigurationError(
699                    "at least one topic is required (or use topic.pattern)".into(),
700                ));
701            }
702            TopicSubscription::Pattern(p) if p.is_empty() => {
703                return Err(ConnectorError::ConfigurationError(
704                    "topic.pattern cannot be empty".into(),
705                ));
706            }
707            _ => {}
708        }
709
710        if self.max_poll_records == 0 {
711            return Err(ConnectorError::ConfigurationError(
712                "max.poll.records must be > 0".into(),
713            ));
714        }
715        if self.reader_channel_capacity < self.max_poll_records {
716            return Err(ConnectorError::ConfigurationError(format!(
717                "reader.channel.capacity ({}) must be >= max.poll.records ({})",
718                self.reader_channel_capacity, self.max_poll_records
719            )));
720        }
721
722        if self.security_protocol.uses_sasl() && self.sasl_mechanism.is_none() {
723            return Err(ConnectorError::ConfigurationError(
724                "sasl.mechanism is required when security.protocol is sasl_plaintext or sasl_ssl"
725                    .into(),
726            ));
727        }
728
729        if let Some(mechanism) = &self.sasl_mechanism {
730            if mechanism.requires_credentials()
731                && (self.sasl_username.is_none() || self.sasl_password.is_none())
732            {
733                return Err(ConnectorError::ConfigurationError(format!(
734                    "sasl.username and sasl.password are required for {mechanism} mechanism"
735                )));
736            }
737        }
738
739        if self.security_protocol.uses_ssl() {
740            if let Some(ref ca) = self.ssl_ca_location {
741                if ca.is_empty() {
742                    return Err(ConnectorError::ConfigurationError(
743                        "ssl.ca.location cannot be empty when specified".into(),
744                    ));
745                }
746            }
747        }
748
749        if self.max_poll_interval.as_secs() < 60 {
750            return Err(ConnectorError::ConfigurationError(format!(
751                "max.poll.interval.ms ({}) must be >= 60000 (60s)",
752                self.max_poll_interval.as_millis()
753            )));
754        }
755
756        // Kafka broker requires heartbeat_interval * 3 < session_timeout.
757        if self.heartbeat_interval.as_millis() * 3 >= self.session_timeout.as_millis() {
758            return Err(ConnectorError::ConfigurationError(format!(
759                "heartbeat.interval.ms ({}) * 3 must be < session.timeout.ms ({})",
760                self.heartbeat_interval.as_millis(),
761                self.session_timeout.as_millis()
762            )));
763        }
764
765        if self.backpressure_high_watermark <= self.backpressure_low_watermark {
766            return Err(ConnectorError::ConfigurationError(
767                "backpressure.high.watermark must be > backpressure.low.watermark".into(),
768            ));
769        }
770        if !(0.0..=1.0).contains(&self.backpressure_high_watermark) {
771            return Err(ConnectorError::ConfigurationError(
772                "backpressure.high.watermark must be between 0.0 and 1.0".into(),
773            ));
774        }
775        if !(0.0..=1.0).contains(&self.backpressure_low_watermark) {
776            return Err(ConnectorError::ConfigurationError(
777                "backpressure.low.watermark must be between 0.0 and 1.0".into(),
778            ));
779        }
780
781        if !(0.0..=1.0).contains(&self.max_deser_error_rate) {
782            return Err(ConnectorError::ConfigurationError(
783                "max.deser.error.rate must be between 0.0 and 1.0".into(),
784            ));
785        }
786
787        if self.format == Format::Avro && self.schema_registry_url.is_none() {
788            return Err(ConnectorError::ConfigurationError(
789                "schema.registry.url is required when format is 'avro'".into(),
790            ));
791        }
792
793        Ok(())
794    }
795
796    /// Builds an rdkafka [`ClientConfig`] from this configuration.
797    #[must_use]
798    pub fn to_rdkafka_config(&self) -> ClientConfig {
799        let mut config = ClientConfig::new();
800
801        config.set("bootstrap.servers", &self.bootstrap_servers);
802        config.set("group.id", &self.group_id);
803        config.set("enable.auto.commit", "false");
804        config.set("enable.auto.offset.store", "false");
805        config.set("auto.offset.reset", self.auto_offset_reset.as_rdkafka_str());
806        config.set(
807            "partition.assignment.strategy",
808            self.partition_assignment_strategy.as_rdkafka_str(),
809        );
810        config.set("security.protocol", self.security_protocol.as_rdkafka_str());
811
812        if let Some(ref mechanism) = self.sasl_mechanism {
813            config.set("sasl.mechanism", mechanism.as_rdkafka_str());
814        }
815        if let Some(ref username) = self.sasl_username {
816            config.set("sasl.username", username);
817        }
818        if let Some(ref password) = self.sasl_password {
819            config.set("sasl.password", password);
820        }
821        if let Some(ref ca) = self.ssl_ca_location {
822            config.set("ssl.ca.location", ca);
823        }
824        if let Some(ref cert) = self.ssl_certificate_location {
825            config.set("ssl.certificate.location", cert);
826        }
827        if let Some(ref key) = self.ssl_key_location {
828            config.set("ssl.key.location", key);
829        }
830        if let Some(ref key_pass) = self.ssl_key_password {
831            config.set("ssl.key.password", key_pass);
832        }
833
834        config.set("isolation.level", self.isolation_level.as_rdkafka_str());
835        config.set(
836            "session.timeout.ms",
837            self.session_timeout.as_millis().to_string(),
838        );
839        config.set(
840            "heartbeat.interval.ms",
841            self.heartbeat_interval.as_millis().to_string(),
842        );
843        config.set(
844            "queued.max.messages.kbytes",
845            self.queued_max_messages_kbytes.to_string(),
846        );
847        config.set(
848            "max.poll.interval.ms",
849            self.max_poll_interval.as_millis().to_string(),
850        );
851
852        if let Some(fetch_min) = self.fetch_min_bytes {
853            config.set("fetch.min.bytes", fetch_min.to_string());
854        }
855
856        if let Some(fetch_max) = self.fetch_max_bytes {
857            config.set("fetch.max.bytes", fetch_max.to_string());
858        }
859
860        if let Some(wait_ms) = self.fetch_max_wait_ms {
861            config.set("fetch.wait.max.ms", wait_ms.to_string());
862        }
863
864        if let Some(partition_max) = self.max_partition_fetch_bytes {
865            config.set("max.partition.fetch.bytes", partition_max.to_string());
866        }
867
868        // Apply pass-through properties, blocking security-critical keys
869        // that could silently downgrade authentication or break semantics.
870        for (key, value) in &self.kafka_properties {
871            if is_blocked_passthrough_key(key) {
872                tracing::warn!(
873                    key,
874                    "ignoring kafka.* pass-through property that overrides a security setting"
875                );
876                continue;
877            }
878            config.set(key, value);
879        }
880
881        config
882    }
883}
884
885/// Returns `true` if a pass-through kafka.* key must not override explicit settings.
886fn is_blocked_passthrough_key(key: &str) -> bool {
887    key.starts_with("sasl.kerberos.")
888        || matches!(
889            key,
890            "security.protocol"
891                | "sasl.mechanism"
892                | "sasl.username"
893                | "sasl.password"
894                | "sasl.oauthbearer.config"
895                | "ssl.ca.location"
896                | "ssl.certificate.location"
897                | "ssl.key.location"
898                | "ssl.key.password"
899                | "ssl.endpoint.identification.algorithm"
900                | "enable.auto.commit"
901                | "enable.auto.offset.store"
902                | "enable.idempotence"
903                | "auto.offset.reset"
904                | "session.timeout.ms"
905                | "heartbeat.interval.ms"
906                | "max.poll.interval.ms"
907                | "queued.max.messages.kbytes"
908                // librdkafka's own auto-commit interval — only meaningful
909                // when `enable.auto.commit=true`, which we hard-disable.
910                | "auto.commit.interval.ms"
911        )
912}
913
914/// Parses a specific offsets string in the format "partition:offset,partition:offset,...".
915///
916/// Example: "0:100,1:200,2:300" maps partition 0 to offset 100, partition 1 to offset 200, etc.
917fn parse_specific_offsets(s: &str) -> Result<HashMap<i32, i64>, ConnectorError> {
918    let mut offsets = HashMap::new();
919
920    for pair in s.split(',') {
921        let pair = pair.trim();
922        if pair.is_empty() {
923            continue;
924        }
925
926        let parts: Vec<&str> = pair.split(':').collect();
927        if parts.len() != 2 {
928            return Err(ConnectorError::ConfigurationError(format!(
929                "invalid offset format '{pair}' (expected 'partition:offset')"
930            )));
931        }
932
933        let partition: i32 = parts[0].trim().parse().map_err(|_| {
934            ConnectorError::ConfigurationError(format!(
935                "invalid partition number '{}' in '{pair}'",
936                parts[0]
937            ))
938        })?;
939
940        let offset: i64 = parts[1].trim().parse().map_err(|_| {
941            ConnectorError::ConfigurationError(format!("invalid offset '{}' in '{pair}'", parts[1]))
942        })?;
943
944        offsets.insert(partition, offset);
945    }
946
947    if offsets.is_empty() {
948        return Err(ConnectorError::ConfigurationError(
949            "startup.specific.offsets cannot be empty".into(),
950        ));
951    }
952
953    Ok(offsets)
954}
955
956#[cfg(test)]
957mod tests;
laminar_connectors/kafka/config/mod.rs

laminar_connectors/kafka/config/
mod.rs
