faucet_source_kinesis/
config.rs

1//! Configuration for the Kinesis source. No I/O here.
2
3use faucet_common_kinesis::KinesisCredentials;
4use schemars::JsonSchema;
5use serde::{Deserialize, Serialize};
6
7/// Where a shard starts reading when no persisted bookmark exists for it.
8///
9/// Serializes as `{ type: <position>, … }` (snake_case discriminators).
10#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, JsonSchema)]
11#[serde(tag = "type", rename_all = "snake_case")]
12pub enum StartPosition {
13    /// Oldest record still in the stream's retention window.
14    #[default]
15    TrimHorizon,
16    /// Only records written after the consumer starts.
17    Latest,
18    /// First record at or after the timestamp (Kinesis matches at **second**
19    /// granularity).
20    AtTimestamp {
21        /// Unix epoch seconds.
22        timestamp_secs: i64,
23    },
24    /// The record with exactly this sequence number.
25    AtSequenceNumber {
26        /// Kinesis sequence number.
27        sequence: String,
28    },
29    /// The record immediately after this sequence number.
30    AfterSequenceNumber {
31        /// Kinesis sequence number.
32        sequence: String,
33    },
34}
35
36/// How each record's `Data` blob is decoded into the emitted `data` field.
37#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
38#[serde(rename_all = "snake_case")]
39pub enum ValueFormat {
40    /// Parse the payload as JSON. A record that is not valid JSON fails the
41    /// stream with a typed error naming the shard + sequence number.
42    #[default]
43    Json,
44    /// Decode the payload as UTF-8 (invalid UTF-8 fails the record).
45    String,
46    /// Base64-encode the raw payload bytes into a JSON string.
47    Bytes,
48}
49
50/// Configuration for [`KinesisSource`](crate::KinesisSource).
51#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
52pub struct KinesisSourceConfig {
53    /// Kinesis Data Stream name.
54    pub stream_name: String,
55    /// AWS region. `None` uses the SDK default chain (env, profile, IMDS).
56    #[serde(default, skip_serializing_if = "Option::is_none")]
57    pub region: Option<String>,
58    /// Custom endpoint URL for LocalStack / VPC endpoints.
59    #[serde(default, skip_serializing_if = "Option::is_none")]
60    pub endpoint_url: Option<String>,
61    /// AWS credentials. Defaults to the SDK default provider chain.
62    #[serde(default)]
63    pub credentials: KinesisCredentials,
64
65    /// Start position for shards with no persisted bookmark.
66    #[serde(default)]
67    pub start_position: StartPosition,
68    /// Explicit shard-id allowlist; empty = every shard.
69    #[serde(default)]
70    pub shard_ids: Vec<String>,
71    /// Also consume closed (post-resharding) shards still inside the
72    /// retention window. Default `false` (open shards only).
73    #[serde(default)]
74    pub include_closed: bool,
75
76    /// Base wait between `GetRecords` calls per shard, in milliseconds.
77    /// Floored at 200 ms (the per-shard 5 reads/sec API limit). Default 1000.
78    #[serde(default = "default_poll_interval_ms")]
79    pub poll_interval_ms: u64,
80    /// `GetRecords` `Limit` per request (1–10000). Default 500.
81    #[serde(default = "default_records_per_request")]
82    pub records_per_request: usize,
83    /// Max shards read concurrently. Default 4.
84    #[serde(default = "default_shard_concurrency")]
85    pub shard_concurrency: usize,
86
87    /// Stop after this many seconds without a new record across all shards.
88    /// At least one of `idle_termination_secs` and `max_messages` must be
89    /// set (mirrors the Kafka source) — a batch run must terminate.
90    #[serde(default, skip_serializing_if = "Option::is_none")]
91    pub idle_termination_secs: Option<u64>,
92    /// Stop after this many records in total.
93    #[serde(default, skip_serializing_if = "Option::is_none")]
94    pub max_messages: Option<usize>,
95
96    /// Payload decoding for the record `data` field.
97    #[serde(default)]
98    pub value_format: ValueFormat,
99
100    /// Records per emitted [`StreamPage`](faucet_core::StreamPage). `0` is
101    /// the "no batching" sentinel: one page per drain cycle. Default 1000.
102    #[serde(default = "default_batch_size")]
103    pub batch_size: usize,
104}
105
106fn default_poll_interval_ms() -> u64 {
107    1000
108}
109fn default_records_per_request() -> usize {
110    500
111}
112fn default_shard_concurrency() -> usize {
113    4
114}
115fn default_batch_size() -> usize {
116    faucet_core::DEFAULT_BATCH_SIZE
117}
118
119impl KinesisSourceConfig {
120    /// Minimal config with defaults for everything but the stream name.
121    pub fn new(stream_name: impl Into<String>) -> Self {
122        Self {
123            stream_name: stream_name.into(),
124            region: None,
125            endpoint_url: None,
126            credentials: KinesisCredentials::default(),
127            start_position: StartPosition::default(),
128            shard_ids: Vec::new(),
129            include_closed: false,
130            poll_interval_ms: default_poll_interval_ms(),
131            records_per_request: default_records_per_request(),
132            shard_concurrency: default_shard_concurrency(),
133            idle_termination_secs: None,
134            max_messages: None,
135            value_format: ValueFormat::default(),
136            batch_size: default_batch_size(),
137        }
138    }
139
140    /// Fail-fast validation, called from `KinesisSource::new`.
141    pub fn validate(&self) -> Result<(), faucet_core::FaucetError> {
142        use faucet_core::FaucetError;
143        if self.stream_name.trim().is_empty() {
144            return Err(FaucetError::Config(
145                "kinesis source: stream_name must not be empty".into(),
146            ));
147        }
148        faucet_core::validate_batch_size(self.batch_size)?;
149        if self.records_per_request == 0 || self.records_per_request > 10_000 {
150            return Err(FaucetError::Config(format!(
151                "kinesis source: records_per_request must be 1..=10000 (got {})",
152                self.records_per_request
153            )));
154        }
155        if self.shard_concurrency == 0 {
156            return Err(FaucetError::Config(
157                "kinesis source: shard_concurrency must be at least 1".into(),
158            ));
159        }
160        if self.idle_termination_secs.is_none() && self.max_messages.is_none() {
161            return Err(FaucetError::Config(
162                "kinesis source: set at least one of idle_termination_secs / max_messages so \
163                 a run can terminate (mirrors the kafka source)"
164                    .into(),
165            ));
166        }
167        if self.idle_termination_secs == Some(0) {
168            return Err(FaucetError::Config(
169                "kinesis source: idle_termination_secs must be at least 1".into(),
170            ));
171        }
172        if self.max_messages == Some(0) {
173            return Err(FaucetError::Config(
174                "kinesis source: max_messages must be at least 1".into(),
175            ));
176        }
177        Ok(())
178    }
179
180    /// Effective per-shard poll interval (floored at 200 ms).
181    pub fn poll_interval(&self) -> std::time::Duration {
182        std::time::Duration::from_millis(self.poll_interval_ms.max(200))
183    }
184}
185
186#[cfg(test)]
187mod tests {
188    use super::*;
189
190    fn valid() -> KinesisSourceConfig {
191        let mut c = KinesisSourceConfig::new("events");
192        c.max_messages = Some(100);
193        c
194    }
195
196    #[test]
197    fn defaults_are_sensible() {
198        let c = KinesisSourceConfig::new("events");
199        assert_eq!(c.start_position, StartPosition::TrimHorizon);
200        assert_eq!(c.value_format, ValueFormat::Json);
201        assert_eq!(c.records_per_request, 500);
202        assert_eq!(c.shard_concurrency, 4);
203        assert_eq!(c.poll_interval_ms, 1000);
204        assert!(!c.include_closed);
205    }
206
207    #[test]
208    fn validation_bounds() {
209        valid().validate().unwrap();
210
211        let mut c = valid();
212        c.stream_name = "  ".into();
213        assert!(c.validate().is_err());
214
215        let mut c = valid();
216        c.records_per_request = 0;
217        assert!(c.validate().is_err());
218        c.records_per_request = 10_001;
219        assert!(c.validate().is_err());
220
221        let mut c = valid();
222        c.shard_concurrency = 0;
223        assert!(c.validate().is_err());
224
225        let mut c = valid();
226        c.max_messages = None;
227        c.idle_termination_secs = None;
228        let err = c.validate().unwrap_err();
229        assert!(err.to_string().contains("idle_termination_secs"), "{err}");
230
231        let mut c = valid();
232        c.idle_termination_secs = Some(0);
233        assert!(c.validate().is_err());
234        let mut c = valid();
235        c.max_messages = Some(0);
236        assert!(c.validate().is_err());
237
238        let mut c = valid();
239        c.batch_size = faucet_core::MAX_BATCH_SIZE + 1;
240        assert!(c.validate().is_err());
241    }
242
243    #[test]
244    fn poll_interval_floored_at_200ms() {
245        let mut c = valid();
246        c.poll_interval_ms = 50;
247        assert_eq!(c.poll_interval(), std::time::Duration::from_millis(200));
248        c.poll_interval_ms = 1500;
249        assert_eq!(c.poll_interval(), std::time::Duration::from_millis(1500));
250    }
251
252    #[test]
253    fn start_positions_parse() {
254        let p: StartPosition = serde_yaml::from_str("type: trim_horizon").unwrap();
255        assert_eq!(p, StartPosition::TrimHorizon);
256        let p: StartPosition = serde_yaml::from_str("type: latest").unwrap();
257        assert_eq!(p, StartPosition::Latest);
258        let p: StartPosition =
259            serde_yaml::from_str("type: at_timestamp\ntimestamp_secs: 1716700000").unwrap();
260        assert_eq!(
261            p,
262            StartPosition::AtTimestamp {
263                timestamp_secs: 1_716_700_000
264            }
265        );
266        let p: StartPosition =
267            serde_yaml::from_str("type: after_sequence_number\nsequence: '49'").unwrap();
268        assert_eq!(
269            p,
270            StartPosition::AfterSequenceNumber {
271                sequence: "49".into()
272            }
273        );
274    }
275
276    #[test]
277    fn full_config_parses_from_yaml() {
278        let yaml = r#"
279stream_name: events
280region: us-east-1
281endpoint_url: http://127.0.0.1:4566
282credentials: { type: access_key, config: { access_key_id: a, secret_access_key: b } }
283start_position: { type: latest }
284shard_ids: ["shardId-000000000000"]
285include_closed: true
286poll_interval_ms: 500
287records_per_request: 1000
288shard_concurrency: 2
289idle_termination_secs: 5
290max_messages: 250
291value_format: string
292batch_size: 100
293"#;
294        let c: KinesisSourceConfig = serde_yaml::from_str(yaml).unwrap();
295        c.validate().unwrap();
296        assert_eq!(c.start_position, StartPosition::Latest);
297        assert_eq!(c.value_format, ValueFormat::String);
298        assert!(c.include_closed);
299    }
300}
faucet_source_kinesis/config.rs

faucet_source_kinesis/
config.rs
