Struct KafkaSourceConfig 

pub struct KafkaSourceConfig {

    pub bootstrap_servers: String,
    pub group_id: String,
    pub subscription: TopicSubscription,
    pub security_protocol: SecurityProtocol,
    pub sasl_mechanism: Option<SaslMechanism>,
    pub sasl_username: Option<String>,
    pub sasl_password: Option<String>,
    pub ssl_ca_location: Option<String>,
    pub ssl_certificate_location: Option<String>,
    pub ssl_key_location: Option<String>,
    pub ssl_key_password: Option<String>,
    pub format: Format,
    pub schema_registry_url: Option<String>,
    pub schema_registry_auth: Option<SrAuth>,
    pub schema_compatibility: Option<CompatibilityLevel>,
    pub schema_registry_ssl_ca_location: Option<String>,
    pub schema_registry_ssl_certificate_location: Option<String>,
    pub schema_registry_ssl_key_location: Option<String>,
    pub event_time_column: Option<String>,
    pub include_metadata: bool,
    pub include_headers: bool,
    pub startup_mode: StartupMode,
    pub auto_offset_reset: OffsetReset,
    pub isolation_level: IsolationLevel,
    pub max_poll_records: usize,
    pub partition_assignment_strategy: AssignmentStrategy,
    pub fetch_min_bytes: Option<i32>,
    pub fetch_max_bytes: Option<i32>,
    pub fetch_max_wait_ms: Option<i32>,
    pub max_partition_fetch_bytes: Option<i32>,
    pub max_out_of_orderness: Duration,
    pub idle_timeout: Duration,
    pub enable_watermark_tracking: bool,
    pub alignment_group_id: Option<String>,
    pub alignment_max_drift: Option<Duration>,
    pub alignment_mode: Option<KafkaAlignmentMode>,
    pub broker_commit_interval: Duration,
    pub backpressure_high_watermark: f64,
    pub backpressure_low_watermark: f64,
    pub kafka_properties: HashMap<String, String>,
}

Kafka source connector configuration.

Uses a custom Debug impl that redacts sasl_password and ssl_key_password to prevent credential leakage in logs.

Fields

bootstrap_servers: String

Comma-separated list of broker addresses.

group_id: String

Consumer group identifier.

subscription: TopicSubscription

Topic subscription (explicit list or regex pattern).

security_protocol: SecurityProtocol

Security protocol for broker connections.

sasl_mechanism: Option<SaslMechanism>

SASL authentication mechanism.

sasl_username: Option<String>

SASL username (for PLAIN, SCRAM-SHA-256, SCRAM-SHA-512).

sasl_password: Option<String>

SASL password (for PLAIN, SCRAM-SHA-256, SCRAM-SHA-512).

ssl_ca_location: Option<String>

Path to SSL CA certificate file (PEM format).

ssl_certificate_location: Option<String>

Path to client SSL certificate file (PEM format).

ssl_key_location: Option<String>

Path to client SSL private key file (PEM format).

ssl_key_password: Option<String>

Password for encrypted SSL private key.

format: Format

Data format for deserialization.

schema_registry_url: Option<String>

Confluent Schema Registry URL.

schema_registry_auth: Option<SrAuth>

Schema Registry authentication credentials.

schema_compatibility: Option<CompatibilityLevel>

Override compatibility level for the subject.

schema_registry_ssl_ca_location: Option<String>

Schema Registry SSL CA certificate path.

schema_registry_ssl_certificate_location: Option<String>

Schema Registry SSL client certificate path.

schema_registry_ssl_key_location: Option<String>

Schema Registry SSL client key path.

event_time_column: Option<String>

Column name containing the event timestamp.

include_metadata: bool

Whether to include Kafka metadata columns (_partition, _offset, _timestamp).

include_headers: bool

Whether to include Kafka headers as a map column (_headers).

startup_mode: StartupMode

Consumer startup mode (controls initial offset positioning).

auto_offset_reset: OffsetReset

Where to start reading when no committed offset exists.

isolation_level: IsolationLevel

Consumer transaction isolation level.

max_poll_records: usize

Maximum records per poll batch.

partition_assignment_strategy: AssignmentStrategy

Partition assignment strategy.

fetch_min_bytes: Option<i32>

Minimum bytes to return from a fetch (allows batching).

fetch_max_bytes: Option<i32>

Maximum bytes to return from broker per request.

fetch_max_wait_ms: Option<i32>

Maximum time broker waits for fetch.min.bytes.

max_partition_fetch_bytes: Option<i32>

Maximum bytes per partition to return from broker.

max_out_of_orderness: Duration

Maximum expected out-of-orderness for watermark generation.

idle_timeout: Duration

Timeout before marking a partition as idle.

enable_watermark_tracking: bool

Enable per-partition watermark tracking (integrates with watermark tracking).

alignment_group_id: Option<String>

Alignment group ID for multi-source coordination (integrates with watermark tracking).

alignment_max_drift: Option<Duration>

Maximum allowed drift between sources in alignment group.

alignment_mode: Option<KafkaAlignmentMode>

Enforcement mode for watermark alignment.

broker_commit_interval: Duration

Interval at which to asynchronously commit offsets to the Kafka broker.

This is advisory — it keeps kafka-consumer-groups lag monitoring accurate. The authoritative offset state lives in LaminarDB’s checkpoint system. Set to Duration::ZERO to disable periodic broker commits (default: 60s).

backpressure_high_watermark: f64

Channel fill ratio at which to pause consumption.

backpressure_low_watermark: f64

Channel fill ratio at which to resume consumption.

kafka_properties: HashMap<String, String>

Additional rdkafka properties passed directly to librdkafka.

Implementations

impl KafkaSourceConfig

pub fn from_config(config: &ConnectorConfig) -> Result<Self, ConnectorError>

Parses a KafkaSourceConfig from a ConnectorConfig.

Errors

Returns ConnectorError if required fields are missing or values are invalid.

pub fn validate(&self) -> Result<(), ConnectorError>

Validates the configuration.

Errors

Returns ConnectorError::ConfigurationError if the configuration is invalid.

pub fn to_rdkafka_config(&self) -> ClientConfig

Builds an rdkafka [ClientConfig] from this configuration.

Trait Implementations

impl Clone for KafkaSourceConfig

fn clone(&self) -> KafkaSourceConfig

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for KafkaSourceConfig

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Default for KafkaSourceConfig

fn default() -> Self

Returns the “default value” for a type.