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

laminar_connectors/kafka/config/
mod.rs
