Struct KafkaSourceConfig 

pub struct KafkaSourceConfig {

    pub bootstrap_servers: Vec<String>,
    pub group_id: String,
    pub topics: Vec<String>,
    pub auto_offset_reset: String,
    pub enable_auto_commit: bool,
    pub auto_commit_interval: Duration,
    pub max_poll_records: usize,
    pub fetch_wait_max: Duration,
    pub fetch_min_bytes: usize,
    pub max_partition_fetch_bytes: usize,
    pub session_timeout: Duration,
    pub heartbeat_interval: Duration,
    pub security_protocol: SecurityProtocol,
    pub sasl_mechanism: Option<SaslMechanism>,
    pub sasl_username: Option<String>,
    pub sasl_password: Option<String>,
    pub ssl_config: Option<SslConfig>,
    pub num_consumer_threads: usize,
    pub queued_max_messages: usize,
    pub socket_receive_buffer_bytes: usize,
    pub isolation_level: String,
    pub additional_properties: HashMap<String, String>,
}

Kafka source (consumer) configuration.

This structure provides all necessary settings for consuming from Kafka topics, including consumer group management, offset handling, and performance tuning.

Examples

use processor::kafka::config::{KafkaSourceConfig, SecurityProtocol};
use std::time::Duration;

let config = KafkaSourceConfig::builder()
    .bootstrap_servers(vec!["localhost:9092".to_string()])
    .group_id("my-consumer-group".to_string())
    .topics(vec!["events".to_string(), "metrics".to_string()])
    .auto_offset_reset("earliest".to_string())
    .enable_auto_commit(false)
    .max_poll_records(500)
    .session_timeout(Duration::from_secs(30))
    .build()
    .expect("valid configuration");

assert!(config.validate().is_ok());

Fields

bootstrap_servers: Vec<String>

Kafka broker addresses (host:port).

group_id: String

Consumer group ID for coordinated consumption.

topics: Vec<String>

Topics to subscribe to.

auto_offset_reset: String

What to do when there is no initial offset: “earliest”, “latest”, “none”.

enable_auto_commit: bool

Enable automatic offset commits.

auto_commit_interval: Duration

Frequency of auto-commits if enabled.

max_poll_records: usize

Maximum number of records to poll in a single batch.

fetch_wait_max: Duration

Maximum time to wait for poll batch to fill.

fetch_min_bytes: usize

Minimum bytes to fetch per request.

max_partition_fetch_bytes: usize

Maximum bytes to fetch per partition.

session_timeout: Duration

Session timeout for consumer group membership.

heartbeat_interval: Duration

Heartbeat interval to coordinator.

security_protocol: SecurityProtocol

Security protocol to use.

sasl_mechanism: Option<SaslMechanism>

SASL mechanism if using SASL authentication.

sasl_username: Option<String>

SASL username.

sasl_password: Option<String>

SASL password.

ssl_config: Option<SslConfig>

SSL configuration.

num_consumer_threads: usize

Number of consumer threads for parallel consumption.

queued_max_messages: usize

Size of the internal queue for buffering messages.

socket_receive_buffer_bytes: usize

Socket receive buffer size.

isolation_level: String

Transaction isolation level: “read_uncommitted” or “read_committed”.

additional_properties: HashMap<String, String>

Additional rdkafka configuration properties.

Implementations

impl KafkaSourceConfig

pub fn builder() -> KafkaSourceConfigBuilder

Create a new builder for KafkaSourceConfig.

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

Validate the configuration.

Errors

Returns an error if:

• Bootstrap servers are empty
• Group ID is empty
• Topics are empty
• Invalid auto_offset_reset value
• Invalid isolation_level value
• SASL credentials missing when SASL is enabled

pub fn to_rdkafka_properties(&self) -> HashMap<String, String>

Convert to rdkafka ClientConfig properties.

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<'de> Deserialize<'de> for KafkaSourceConfig

fn deserialize<__D>(__deserializer: __D) -> Result<Self, __D::Error>

where __D: Deserializer<'de>,

Deserialize this value from the given Serde deserializer.

impl Serialize for KafkaSourceConfig

fn serialize<__S>(&self, __serializer: __S) -> Result<__S::Ok, __S::Error>

where __S: Serializer,

Serialize this value into the given Serde serializer.