rust-data-processing 0.3.3

Schema-first ingestion (CSV, JSON, Parquet, Excel) into an in-memory DataSet, plus Polars-backed pipelines, SQL, profiling, validation, and map/reduce-style processing.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131

//! Builder-style configuration for native Kafka consumer/producer paths.

use crate::error::{IngestionError, IngestionResult};

/// Consumer configuration (brokers, group, topic, poll limits).
#[derive(Debug, Clone)]
pub struct KafkaConsumerBuilder {
    pub brokers: String,
    pub group_id: String,
    pub topic: String,
    pub auto_offset_reset: String,
    pub session_timeout_ms: Option<u32>,
}

impl KafkaConsumerBuilder {
    pub fn new(
        brokers: impl Into<String>,
        group_id: impl Into<String>,
        topic: impl Into<String>,
    ) -> Self {
        Self {
            brokers: brokers.into(),
            group_id: group_id.into(),
            topic: topic.into(),
            auto_offset_reset: "earliest".to_string(),
            session_timeout_ms: None,
        }
    }

    pub fn auto_offset_reset(mut self, value: impl Into<String>) -> Self {
        self.auto_offset_reset = value.into();
        self
    }

    pub fn session_timeout_ms(mut self, ms: u32) -> Self {
        self.session_timeout_ms = Some(ms);
        self
    }

    pub fn validate(&self) -> IngestionResult<()> {
        if self.brokers.trim().is_empty() {
            return Err(IngestionError::SchemaMismatch {
                message: "kafka brokers must not be empty".to_string(),
            });
        }
        if self.group_id.trim().is_empty() {
            return Err(IngestionError::SchemaMismatch {
                message: "kafka group_id must not be empty".to_string(),
            });
        }
        if self.topic.trim().is_empty() {
            return Err(IngestionError::SchemaMismatch {
                message: "kafka topic must not be empty".to_string(),
            });
        }
        Ok(())
    }
}

/// Producer configuration (brokers, topic, optional key/value columns).
#[derive(Debug, Clone, Default)]
pub struct KafkaProducerBuilder {
    pub brokers: String,
    pub topic: String,
    pub message_timeout_ms: u64,
    pub key_column: Option<String>,
    pub value_column: Option<String>,
    pub headers: Vec<(String, String)>,
}

impl KafkaProducerBuilder {
    pub fn new(brokers: impl Into<String>, topic: impl Into<String>) -> Self {
        Self {
            brokers: brokers.into(),
            topic: topic.into(),
            message_timeout_ms: 5_000,
            ..Self::default()
        }
    }

    pub fn message_timeout_ms(mut self, ms: u64) -> Self {
        self.message_timeout_ms = ms;
        self
    }

    pub fn key_column(mut self, column: impl Into<String>) -> Self {
        self.key_column = Some(column.into());
        self
    }

    pub fn value_column(mut self, column: impl Into<String>) -> Self {
        self.value_column = Some(column.into());
        self
    }

    pub fn header(mut self, name: impl Into<String>, value: impl Into<String>) -> Self {
        self.headers.push((name.into(), value.into()));
        self
    }

    pub fn validate(&self) -> IngestionResult<()> {
        if self.brokers.trim().is_empty() {
            return Err(IngestionError::SchemaMismatch {
                message: "kafka brokers must not be empty".to_string(),
            });
        }
        if self.topic.trim().is_empty() {
            return Err(IngestionError::SchemaMismatch {
                message: "kafka topic must not be empty".to_string(),
            });
        }
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn consumer_builder_rejects_empty_brokers() {
        let b = KafkaConsumerBuilder::new("", "g", "t");
        assert!(b.validate().is_err());
    }

    #[test]
    fn producer_builder_accepts_minimal_config() {
        let b = KafkaProducerBuilder::new("localhost:9092", "out");
        assert!(b.validate().is_ok());
    }
}