Struct KafkaSinkConfig 

pub struct KafkaSinkConfig {

    pub bootstrap_servers: Vec<String>,
    pub topic: String,
    pub acks: String,
    pub retries: u32,
    pub max_in_flight_requests: usize,
    pub enable_idempotence: bool,
    pub request_timeout: Duration,
    pub batch_size: usize,
    pub linger: Duration,
    pub buffer_memory: usize,
    pub max_block: Duration,
    pub compression: CompressionType,
    pub max_request_size: usize,
    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 socket_send_buffer_bytes: usize,
    pub queued_max_messages: usize,
    pub partitioner: Option<String>,
    pub additional_properties: HashMap<String, String>,
}

Kafka sink (producer) configuration.

This structure provides all necessary settings for producing to Kafka topics, including delivery guarantees, batching, compression, and performance tuning.

Examples

use processor::kafka::config::{KafkaSinkConfig, CompressionType};
use std::time::Duration;

let config = KafkaSinkConfig::builder()
    .bootstrap_servers(vec!["localhost:9092".to_string()])
    .topic("processed-events".to_string())
    .compression(CompressionType::Lz4)
    .batch_size(16384)
    .linger(Duration::from_millis(10))
    .acks("all".to_string())
    .max_in_flight_requests(5)
    .enable_idempotence(true)
    .build()
    .expect("valid configuration");

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

Fields

bootstrap_servers: Vec<String>

Kafka broker addresses (host:port).

topic: String

Default topic to produce to.

acks: String

Number of acknowledgments required: “0”, “1”, or “all”.

retries: u32

Number of retries for transient errors.

max_in_flight_requests: usize

Maximum in-flight requests per connection.

enable_idempotence: bool

Enable idempotent producer.

request_timeout: Duration

Request timeout.

batch_size: usize

Batch size in bytes.

linger: Duration

Time to wait before sending a batch (even if not full).

buffer_memory: usize

Total memory buffer for producer.

max_block: Duration

Maximum time to block on send.

compression: CompressionType

Compression type for messages.

max_request_size: usize

Maximum size of a single message.

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.

socket_send_buffer_bytes: usize

Socket send buffer size.

queued_max_messages: usize

Size of the internal queue for buffering messages.

partitioner: Option<String>

Custom partitioner class (rdkafka specific).

additional_properties: HashMap<String, String>

Additional rdkafka configuration properties.

Implementations

impl KafkaSinkConfig

pub fn builder() -> KafkaSinkConfigBuilder

Create a new builder for KafkaSinkConfig.

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

Validate the configuration.

Errors

Returns an error if:

• Bootstrap servers are empty
• Topic is empty
• Invalid acks 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 KafkaSinkConfig

fn clone(&self) -> Self

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl Debug for KafkaSinkConfig

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

Formats the value using the given formatter.

impl<'de> Deserialize<'de> for KafkaSinkConfig

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

where __D: Deserializer<'de>,

Deserialize this value from the given Serde deserializer.

impl Serialize for KafkaSinkConfig

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

where __S: Serializer,

Serialize this value into the given Serde serializer.