spate_avro/
config.rs

1//! Configuration and construction.
2//!
3//! The `deserializer: { avro: { ... } }` section of the pipeline YAML is
4//! handed here as an opaque
5//! [`ComponentConfig`](spate_core::config::ComponentConfig):
6//!
7//! ```yaml
8//! deserializer:
9//!   avro:
10//!     mode: confluent                # confluent | raw | single_object
11//!     registry:                      # required for confluent
12//!       url: ${SCHEMA_REGISTRY_URL}
13//!       username: ${SR_USER}         # optional basic auth
14//!       password: ${SR_PASSWORD}
15//!     prewarm_subjects: [orders-value]
16//!     negative_cache_ttl: 30s
17//!     reader_schema:                 # optional: pin the resolved shape
18//!       path: /etc/spate/orders.avsc   # (or `inline: '{"type": ...}'`)
19//!     # schema: { inline | path }    # required for raw / single_object
20//! ```
21
22use crate::cache::CompiledSchema;
23use crate::deser::{AvroSerdeDeserializer, AvroValueDeserializer, DecoderCore, SchemaSourceMode};
24use crate::registry::{RegistryConfig, spawn_fetcher};
25use apache_avro::Schema;
26use apache_avro::rabin::Rabin;
27use serde::Deserialize;
28use spate_core::config::ComponentConfig;
29use std::sync::Arc;
30use std::time::Duration;
31
32/// Payload framing mode.
33#[derive(Clone, Copy, Debug, Default, Deserialize, PartialEq, Eq)]
34#[serde(rename_all = "snake_case")]
35pub enum AvroMode {
36    /// Confluent wire format with a schema registry.
37    #[default]
38    Confluent,
39    /// Bare datums with a fixed schema.
40    Raw,
41    /// Avro single-object encoding with a fixed schema.
42    SingleObject,
43}
44
45/// A schema provided inline or from a file — set exactly one field.
46#[derive(Clone, Debug, Default, Deserialize)]
47#[serde(deny_unknown_fields, default)]
48pub struct SchemaSource {
49    /// The schema JSON itself.
50    pub inline: Option<String>,
51    /// Path to a `.avsc` file (Kubernetes: a mounted ConfigMap).
52    pub path: Option<std::path::PathBuf>,
53}
54
55impl SchemaSource {
56    /// An inline schema.
57    #[must_use]
58    pub fn inline(schema: impl Into<String>) -> Self {
59        SchemaSource {
60            inline: Some(schema.into()),
61            path: None,
62        }
63    }
64
65    /// A schema loaded from a file.
66    #[must_use]
67    pub fn path(path: impl Into<std::path::PathBuf>) -> Self {
68        SchemaSource {
69            inline: None,
70            path: Some(path.into()),
71        }
72    }
73
74    /// Load the schema's original JSON source (backend compiles need the
75    /// source text, not a re-rendered canonical form).
76    fn load_text(&self) -> Result<String, AvroConfigError> {
77        match (&self.inline, &self.path) {
78            (Some(s), None) => Ok(s.clone()),
79            (None, Some(p)) => {
80                std::fs::read_to_string(p).map_err(|e| AvroConfigError::SchemaLoad {
81                    detail: format!("{}: {e}", p.display()),
82                })
83            }
84            _ => Err(AvroConfigError::Invalid {
85                detail: "a schema source needs exactly one of `inline` or `path`".into(),
86            }),
87        }
88    }
89
90    /// Load and parse the schema with `apache-avro` (the reader-schema
91    /// path, which only exists on the apache backend).
92    fn load(&self) -> Result<Schema, AvroConfigError> {
93        let text = self.load_text()?;
94        Schema::parse_str(&text).map_err(|e| AvroConfigError::SchemaLoad {
95            detail: format!("schema failed to parse: {e}"),
96        })
97    }
98}
99
100/// Registry connection section.
101#[derive(Clone, Debug, Deserialize)]
102#[serde(deny_unknown_fields)]
103pub struct RegistrySection {
104    /// Base URL of the Confluent-compatible schema registry.
105    pub url: String,
106    /// Basic-auth username (optional).
107    #[serde(default)]
108    pub username: Option<String>,
109    /// Basic-auth password (optional; use `${VAR}` interpolation).
110    #[serde(default)]
111    pub password: Option<String>,
112}
113
114/// The `avro` component configuration.
115#[derive(Clone, Debug, Deserialize)]
116#[serde(deny_unknown_fields, default)]
117pub struct AvroSettings {
118    /// Payload framing mode.
119    pub mode: AvroMode,
120    /// Registry connection (required in `confluent` mode).
121    pub registry: Option<RegistrySection>,
122    /// Subjects whose latest schemas are fetched at startup.
123    pub prewarm_subjects: Vec<String>,
124    /// How long a failed schema id stays negatively cached before a
125    /// refetch is allowed.
126    #[serde(with = "humantime_serde")]
127    pub negative_cache_ttl: Duration,
128    /// Optional reader schema pinning the resolved record shape
129    /// (`confluent` and `raw` modes).
130    pub reader_schema: Option<SchemaSource>,
131    /// The writer schema (required in `raw` and `single_object` modes).
132    pub schema: Option<SchemaSource>,
133}
134
135impl Default for AvroSettings {
136    fn default() -> Self {
137        AvroSettings {
138            mode: AvroMode::Confluent,
139            registry: None,
140            prewarm_subjects: Vec::new(),
141            negative_cache_ttl: Duration::from_secs(30),
142            reader_schema: None,
143            schema: None,
144        }
145    }
146}
147
148/// Construction errors.
149#[derive(Debug, thiserror::Error)]
150#[non_exhaustive]
151pub enum AvroConfigError {
152    /// The component section did not deserialize.
153    #[error(transparent)]
154    Config(#[from] spate_core::config::ConfigError),
155    /// A configured schema could not be loaded or parsed.
156    #[error("avro schema: {detail}")]
157    SchemaLoad {
158        /// What went wrong.
159        detail: String,
160    },
161    /// The settings are inconsistent for the chosen mode.
162    #[error("avro configuration: {detail}")]
163    Invalid {
164        /// What went wrong.
165        detail: String,
166    },
167}
168
169/// Builder produced from the opaque config section; hands out either the
170/// dynamically-typed or the serde-typed deserializer.
171#[derive(Clone, Debug)]
172pub struct AvroDeserializerBuilder {
173    core: DecoderCore,
174}
175
176impl AvroDeserializerBuilder {
177    /// Build from the pipeline's `deserializer` component section.
178    ///
179    /// `runtime` hosts the registry fetcher task and the startup pre-warm
180    /// (Confluent mode); pass the pipeline's I/O runtime handle.
181    pub fn from_component(
182        cfg: &ComponentConfig,
183        runtime: &tokio::runtime::Handle,
184    ) -> Result<Self, AvroConfigError> {
185        let settings: AvroSettings = cfg.deserialize_into()?;
186        Self::from_settings(&settings, runtime)
187    }
188
189    /// Build from already-parsed settings.
190    pub fn from_settings(
191        settings: &AvroSettings,
192        runtime: &tokio::runtime::Handle,
193    ) -> Result<Self, AvroConfigError> {
194        let reader_schema = settings
195            .reader_schema
196            .as_ref()
197            .map(|s| s.load().map(Arc::new))
198            .transpose()?;
199        let mode = match settings.mode {
200            AvroMode::Confluent => {
201                let registry = settings.registry.as_ref().ok_or(AvroConfigError::Invalid {
202                    detail: "mode `confluent` requires a `registry` section".into(),
203                })?;
204                if settings.schema.is_some() {
205                    return Err(AvroConfigError::Invalid {
206                        detail: "`schema` is only used in `raw`/`single_object` modes; \
207                                 in `confluent` mode writer schemas come from the registry \
208                                 (use `reader_schema` to pin the resolved shape)"
209                            .into(),
210                    });
211                }
212                let registry_cfg = RegistryConfig {
213                    url: registry.url.clone(),
214                    basic_auth: registry
215                        .username
216                        .as_ref()
217                        .map(|u| (u.clone(), registry.password.clone())),
218                    negative_cache_ttl: settings.negative_cache_ttl,
219                };
220                let handle = spawn_fetcher(registry_cfg.clone(), runtime);
221                if !settings.prewarm_subjects.is_empty() {
222                    let subjects = settings.prewarm_subjects.clone();
223                    let cache = Arc::clone(&handle.cache);
224                    runtime.spawn(async move {
225                        crate::registry::prewarm(&registry_cfg, &subjects, &cache).await;
226                    });
227                }
228                SchemaSourceMode::Confluent {
229                    registry: handle,
230                    memo: crate::cache::SchemaCache::empty_snapshot(),
231                }
232            }
233            AvroMode::Raw => {
234                let schema = Self::fixed_schema(settings, "raw")?;
235                if !settings.prewarm_subjects.is_empty() {
236                    return Err(AvroConfigError::Invalid {
237                        detail: "`prewarm_subjects` requires mode `confluent`".into(),
238                    });
239                }
240                SchemaSourceMode::Raw { schema }
241            }
242            AvroMode::SingleObject => {
243                let schema = Self::fixed_schema(settings, "single_object")?;
244                if !settings.prewarm_subjects.is_empty() {
245                    return Err(AvroConfigError::Invalid {
246                        detail: "`prewarm_subjects` requires mode `confluent`".into(),
247                    });
248                }
249                if settings.reader_schema.is_some() {
250                    return Err(AvroConfigError::Invalid {
251                        detail: "single_object mode decodes with the configured schema \
252                                 directly; `reader_schema` is not supported"
253                            .into(),
254                    });
255                }
256                // The single-object header fingerprint is computed from the
257                // parsed schema, so this mode needs the parse to succeed at
258                // settings-load time (unlike `raw`).
259                let apache =
260                    schema
261                        .schema
262                        .as_ref()
263                        .map_err(|reason| AvroConfigError::SchemaLoad {
264                            detail: format!(
265                                "mode `single_object` computes the header fingerprint \
266                                 with the apache parser, which rejected the schema: {reason}"
267                            ),
268                        })?;
269                let fp = apache.fingerprint::<Rabin>();
270                let bytes: [u8; 8] =
271                    fp.bytes
272                        .as_slice()
273                        .try_into()
274                        .map_err(|_| AvroConfigError::Invalid {
275                            detail: "unexpected Rabin fingerprint width".into(),
276                        })?;
277                SchemaSourceMode::SingleObject {
278                    schema,
279                    fingerprint: u64::from_le_bytes(bytes),
280                }
281            }
282        };
283        Ok(AvroDeserializerBuilder {
284            core: DecoderCore {
285                mode,
286                reader_schema,
287            },
288        })
289    }
290
291    fn fixed_schema(
292        settings: &AvroSettings,
293        mode: &str,
294    ) -> Result<Arc<CompiledSchema>, AvroConfigError> {
295        let source = settings.schema.as_ref().ok_or(AvroConfigError::Invalid {
296            detail: format!("mode `{mode}` requires a `schema` (inline or path)"),
297        })?;
298        if settings.registry.is_some() {
299            return Err(AvroConfigError::Invalid {
300                detail: format!("mode `{mode}` does not use a `registry` section"),
301            });
302        }
303        let json = source.load_text()?;
304        // Per-backend compile (with apache-avro's parse-panic caught): a
305        // fixed schema only one backend accepts still builds, and each
306        // backend's builder gates on its own side below. Only a schema no
307        // enabled backend accepts is a load error.
308        let compiled = CompiledSchema::compile(0, &json);
309        if let Some(reason) = compiled.unusable_reason() {
310            return Err(AvroConfigError::SchemaLoad { detail: reason });
311        }
312        Ok(Arc::new(compiled))
313    }
314
315    /// The fixed schema's compile failure, if any. Fixed-schema modes
316    /// compile eagerly at build time, so an unusable schema fails the
317    /// builders *here* rather than surfacing per record as
318    /// `SchemaUnavailable` — which under the default Skip policy would drop
319    /// and ack 100% of the input while watermarks advance. Confluent writer
320    /// schemas arrive per id at runtime and cannot be checked until then.
321    fn fixed_schema_error(&self) -> Option<String> {
322        match &self.core.mode {
323            SchemaSourceMode::Raw { schema } | SchemaSourceMode::SingleObject { schema, .. } => {
324                schema.schema.as_ref().err().cloned()
325            }
326            SchemaSourceMode::Confluent { .. } => None,
327        }
328    }
329
330    /// The dynamically-typed deserializer (emits [`crate::AvroValue`]).
331    ///
332    /// # Errors
333    ///
334    /// Rejects a fixed schema (`raw`/`single_object`) that cannot be
335    /// parsed — deferring it would surface every record as
336    /// `SchemaUnavailable` and drop it under the default Skip policy.
337    pub fn build_value(&self) -> Result<AvroValueDeserializer, AvroConfigError> {
338        if let Some(reason) = self.fixed_schema_error() {
339            return Err(AvroConfigError::Invalid { detail: reason });
340        }
341        Ok(AvroValueDeserializer::new(self.core.clone()))
342    }
343
344    /// The serde-typed deserializer (emits `T`).
345    ///
346    /// # Errors
347    ///
348    /// Rejects a fixed schema that cannot be parsed — see
349    /// [`Self::build_value`].
350    pub fn build_serde<T>(&self) -> Result<AvroSerdeDeserializer<T>, AvroConfigError>
351    where
352        T: serde::de::DeserializeOwned + Send + 'static,
353    {
354        if let Some(reason) = self.fixed_schema_error() {
355            return Err(AvroConfigError::Invalid { detail: reason });
356        }
357        Ok(AvroSerdeDeserializer::new(self.core.clone()))
358    }
359}
360
361#[cfg(test)]
362mod tests {
363    use super::*;
364    use spate_core::config::ComponentConfig;
365
366    const SCHEMA: &str = r#"{"type":"record","name":"E","fields":[{"name":"id","type":"long"}]}"#;
367
368    fn component(yaml: &str) -> ComponentConfig {
369        ComponentConfig::new("avro", serde_yaml::from_str(yaml).unwrap())
370    }
371
372    fn runtime() -> tokio::runtime::Runtime {
373        tokio::runtime::Builder::new_current_thread()
374            .enable_all()
375            .build()
376            .unwrap()
377    }
378
379    #[test]
380    fn full_confluent_section_parses() {
381        let cfg = component(
382            r#"
383            mode: confluent
384            registry:
385              url: http://sr:8081
386              username: svc
387              password: secret
388            prewarm_subjects: [orders-value, users-value]
389            negative_cache_ttl: 45s
390            reader_schema:
391              inline: '{"type":"record","name":"E","fields":[{"name":"id","type":"long"}]}'
392            "#,
393        );
394        let settings: AvroSettings = cfg.deserialize_into().unwrap();
395        assert_eq!(settings.mode, AvroMode::Confluent);
396        assert_eq!(settings.prewarm_subjects.len(), 2);
397        assert_eq!(settings.negative_cache_ttl, Duration::from_secs(45));
398        let rt = runtime();
399        AvroDeserializerBuilder::from_settings(&settings, rt.handle()).unwrap();
400    }
401
402    #[test]
403    fn unknown_fields_are_rejected() {
404        let cfg = component("mode: raw\nschema: {inline: '\"long\"'}\nturbo: true\n");
405        let err = cfg.deserialize_into::<AvroSettings>().unwrap_err();
406        assert!(err.to_string().contains("turbo"), "{err}");
407    }
408
409    #[test]
410    fn mode_requirements_are_validated() {
411        let rt = runtime();
412        let cases: &[(&str, &str)] = &[
413            ("mode: confluent", "requires a `registry`"),
414            ("mode: raw", "requires a `schema`"),
415            ("mode: single_object", "requires a `schema`"),
416        ];
417        for (yaml, needle) in cases {
418            let settings: AvroSettings = component(yaml).deserialize_into().unwrap();
419            let err = AvroDeserializerBuilder::from_settings(&settings, rt.handle()).unwrap_err();
420            assert!(err.to_string().contains(needle), "{yaml}: {err}");
421        }
422    }
423
424    #[test]
425    fn conflicting_sections_are_rejected() {
426        let rt = runtime();
427        let confluent_with_schema = format!(
428            "mode: confluent\nregistry: {{url: http://sr}}\nschema: {{inline: '{SCHEMA}'}}"
429        );
430        let raw_with_registry =
431            format!("mode: raw\nschema: {{inline: '{SCHEMA}'}}\nregistry: {{url: http://sr}}");
432        let raw_with_prewarm =
433            format!("mode: raw\nschema: {{inline: '{SCHEMA}'}}\nprewarm_subjects: [x]");
434        let so_with_prewarm =
435            format!("mode: single_object\nschema: {{inline: '{SCHEMA}'}}\nprewarm_subjects: [x]");
436        let so_with_reader = format!(
437            "mode: single_object\nschema: {{inline: '{SCHEMA}'}}\nreader_schema: {{inline: '{SCHEMA}'}}"
438        );
439        for yaml in [
440            confluent_with_schema,
441            raw_with_registry,
442            raw_with_prewarm,
443            so_with_prewarm,
444            so_with_reader,
445        ] {
446            let settings: AvroSettings = component(&yaml).deserialize_into().unwrap();
447            assert!(
448                AvroDeserializerBuilder::from_settings(&settings, rt.handle()).is_err(),
449                "must reject: {yaml}"
450            );
451        }
452    }
453
454    #[test]
455    fn schema_from_file_and_bad_schema_errors() {
456        let rt = runtime();
457        let dir = tempfile::tempdir().unwrap();
458        let path = dir.path().join("e.avsc");
459        std::fs::write(&path, SCHEMA).unwrap();
460        let settings: AvroSettings =
461            component(&format!("mode: raw\nschema: {{path: {}}}", path.display()))
462                .deserialize_into()
463                .unwrap();
464        AvroDeserializerBuilder::from_settings(&settings, rt.handle()).unwrap();
465
466        let settings: AvroSettings = component("mode: raw\nschema: {inline: 'not json'}")
467            .deserialize_into()
468            .unwrap();
469        let err = AvroDeserializerBuilder::from_settings(&settings, rt.handle()).unwrap_err();
470        assert!(matches!(err, AvroConfigError::SchemaLoad { .. }), "{err}");
471    }
472}
spate_avro/config.rs

spate_avro/
config.rs
