etl_core/config/
mod.rs

1//! Pipeline configuration: typed framework sections plus opaque
2//! per-component passthrough, loaded from YAML with `${VAR:-default}`
3//! environment interpolation.
4//!
5//! The framework owns the typed sections (`pipeline`, `checkpoint`,
6//! `backpressure`, `metrics`) and validates them strictly
7//! (`deny_unknown_fields` at every level). The `source`, `deserializer`,
8//! and `sink` sections are single-key mappings selecting a component type;
9//! their bodies are opaque [`ComponentConfig`]s handed to the component's
10//! factory, which deserializes its own typed config. This keeps connectors
11//! fully decoupled from the framework schema.
12//!
13//! ```yaml
14//! pipeline: { name: orders, threads: 4, io_threads: 2 }
15//! checkpoint: { interval: 5s, max_pending_batches: 1024 }
16//! backpressure: { max_inflight_bytes: 256MiB }
17//! source:
18//!   kafka:                                   # KafkaSourceConfig
19//!     brokers: ${KAFKA_BROKERS:-localhost:9092}
20//!     topic: orders
21//!     group_id: orders-etl                   # required (no default)
22//! deserializer:
23//!   avro:                                    # AvroSettings (confluent mode)
24//!     registry:
25//!       url: ${SCHEMA_REGISTRY_URL:?schema registry required}
26//! sink:
27//!   clickhouse:                              # ClickHouseSinkConfig
28//!     table: orders_local
29//!     columns: [id, amount, ts]              # required; order is the wire contract
30//!     shards:
31//!       - { replicas: ["http://ch-0-0:8123", "http://ch-0-1:8123"] }
32//! metrics: { exporter: prometheus, listen: 0.0.0.0:9090 }
33//! ```
34//!
35//! Environment interpolation runs on the raw text before parsing — see
36//! [`interpolate`](self::interpolate::interpolate_with) for the exact
37//! semantics of `${VAR}`, `${VAR:-default}`, `${VAR:?message}`, and `$$`.
38//!
39//! # Example
40//!
41//! ```
42//! use etl_core::config::PipelineConfig;
43//!
44//! let cfg = PipelineConfig::from_str(r#"
45//! pipeline: { name: demo }
46//! source: { memory: {} }
47//! sink: { memory: {} }
48//! "#).unwrap();
49//!
50//! assert_eq!(cfg.pipeline.name, "demo");
51//! assert_eq!(cfg.pipeline.io_threads, 2);                    // default
52//! assert_eq!(cfg.checkpoint.max_pending_batches, 1024);      // default
53//! assert_eq!(cfg.source.type_tag(), "memory");
54//! ```
55
56mod component;
57mod error;
58mod interpolate;
59
60pub use component::ComponentConfig;
61pub use error::ConfigError;
62
63/// Re-export of `serde_yaml::Value`, the opaque body type carried by a
64/// [`ComponentConfig`]. `serde_yaml` is a 0.x dependency, so exposing its
65/// `Value` directly in `etl-core`'s public API would tie our semver to
66/// theirs; this alias is the documented exemption (mirroring the [`bytes`]
67/// and `AvroValue` re-export pattern — see `docs/DESIGN.md` § Dependency
68/// policy). A major bump of the YAML crate becomes a breaking change here,
69/// and only here.
70///
71/// [`bytes`]: crate::bytes
72pub use serde_yaml::Value as YamlValue;
73
74use bytesize::ByteSize;
75use serde::Deserialize;
76use std::net::SocketAddr;
77use std::path::Path;
78use std::time::Duration;
79
80/// Root of a pipeline's configuration file.
81///
82/// One process runs one pipeline; one file configures one process.
83#[derive(Debug, PartialEq, Deserialize)]
84#[serde(deny_unknown_fields)]
85pub struct PipelineConfig {
86    /// Identity and thread budget.
87    pub pipeline: PipelineSection,
88    /// Watermark commit policy.
89    #[serde(default)]
90    pub checkpoint: CheckpointSection,
91    /// In-flight budget and pause/resume hysteresis.
92    #[serde(default)]
93    pub backpressure: BackpressureSection,
94    /// Exporter selection and observability knobs.
95    #[serde(default)]
96    pub metrics: MetricsSection,
97    /// The source component (opaque body).
98    pub source: ComponentConfig,
99    /// Optional deserializer component (opaque body). Sources that emit
100    /// ready-made records need none.
101    #[serde(default)]
102    pub deserializer: Option<ComponentConfig>,
103    /// The sink component (opaque body).
104    pub sink: ComponentConfig,
105}
106
107/// `pipeline:` — identity and thread budget.
108#[derive(Debug, PartialEq, Deserialize)]
109#[serde(deny_unknown_fields)]
110pub struct PipelineSection {
111    /// Pipeline name; the `pipeline` label on every metric.
112    pub name: String,
113    /// Pinned pipeline thread count. `None` derives from
114    /// `available_parallelism` minus the I/O reserve at startup.
115    #[serde(default)]
116    pub threads: Option<usize>,
117    /// Worker threads for the I/O runtime (sink workers, checkpointer,
118    /// admin server).
119    #[serde(default = "defaults::io_threads")]
120    pub io_threads: usize,
121    /// Core-pinning mode for pipeline threads.
122    #[serde(default)]
123    pub pinning: PinningMode,
124}
125
126/// How pipeline threads are pinned to cores.
127#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize, Default)]
128#[serde(rename_all = "snake_case")]
129#[non_exhaustive]
130pub enum PinningMode {
131    /// No pinning (default). Correct choice unless the pod has exclusive
132    /// cores (Kubernetes static CPU manager + Guaranteed QoS).
133    #[default]
134    Off,
135    /// Pin thread *i* to core *i*.
136    Compact,
137}
138
139/// `checkpoint:` — watermark commit policy.
140#[derive(Debug, PartialEq, Deserialize)]
141#[serde(deny_unknown_fields, default)]
142pub struct CheckpointSection {
143    /// How often committable watermarks are flushed to the source.
144    #[serde(with = "humantime_serde")]
145    pub interval: Duration,
146    /// Maximum unacknowledged batches per partition before that partition
147    /// is paused (doubles as a backpressure trigger).
148    pub max_pending_batches: usize,
149    /// Shutdown/rebalance drain budget. Must be comfortably below the pod's
150    /// `terminationGracePeriodSeconds`.
151    #[serde(with = "humantime_serde")]
152    pub drain_timeout: Duration,
153    /// A partition watermark stalled behind a failed batch for longer than
154    /// this fails the pipeline. Failed batches only stall watermarks
155    /// permanently (their data replays after restart), so this converts a
156    /// permanent sink failure — a dropped table, revoked credentials — into
157    /// a clean `Failed` exit and a restart instead of a process that runs on
158    /// forever, consuming the source but committing nothing for that
159    /// partition.
160    #[serde(with = "humantime_serde")]
161    pub stalled_fail_after: Duration,
162}
163
164impl Default for CheckpointSection {
165    fn default() -> Self {
166        CheckpointSection {
167            interval: Duration::from_secs(5),
168            max_pending_batches: 1024,
169            drain_timeout: Duration::from_secs(25),
170            stalled_fail_after: Duration::from_secs(120),
171        }
172    }
173}
174
175/// `backpressure:` — in-flight budget and hysteresis.
176#[derive(Debug, PartialEq, Deserialize)]
177#[serde(deny_unknown_fields, default)]
178pub struct BackpressureSection {
179    /// Global cap on bytes admitted into the pipeline but not yet durably
180    /// written.
181    pub max_inflight_bytes: ByteSize,
182    /// Fraction of the budget at which sources are paused.
183    pub high_ratio: f64,
184    /// Fraction of the budget below which sources may resume.
185    pub low_ratio: f64,
186    /// Minimum pause duration before resuming (avoids flapping; pausing a
187    /// Kafka partition purges its prefetch, so resume is not free).
188    #[serde(with = "humantime_serde")]
189    pub min_pause: Duration,
190}
191
192impl Default for BackpressureSection {
193    fn default() -> Self {
194        BackpressureSection {
195            max_inflight_bytes: ByteSize::mib(256),
196            high_ratio: 0.8,
197            low_ratio: 0.5,
198            min_pause: Duration::from_millis(500),
199        }
200    }
201}
202
203/// `metrics:` — exporter selection and observability knobs.
204#[derive(Debug, PartialEq, Deserialize)]
205#[serde(deny_unknown_fields, default)]
206pub struct MetricsSection {
207    /// Which exporter to install.
208    pub exporter: MetricsExporter,
209    /// Admin server bind address (`/metrics`, `/healthz`, `/readyz`).
210    pub listen: SocketAddr,
211    /// Emit per-partition gauge series (`partition` label). Off by default:
212    /// cardinality grows with the assignment.
213    pub per_partition_detail: bool,
214    /// Time basis for `etl_e2e_latency_seconds`.
215    pub e2e_basis: E2eBasis,
216}
217
218impl Default for MetricsSection {
219    fn default() -> Self {
220        MetricsSection {
221            exporter: MetricsExporter::Prometheus,
222            listen: SocketAddr::from(([0, 0, 0, 0], 9090)),
223            per_partition_detail: false,
224            e2e_basis: E2eBasis::Ingest,
225        }
226    }
227}
228
229/// Metrics exporter selection.
230#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize, Default)]
231#[serde(rename_all = "snake_case")]
232#[non_exhaustive]
233pub enum MetricsExporter {
234    /// Prometheus scrape endpoint on the admin server (default).
235    #[default]
236    Prometheus,
237    /// No exporter (metrics recorded to a no-op recorder).
238    None,
239}
240
241/// Which timestamp anchors end-to-end latency.
242#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize, Default)]
243#[serde(rename_all = "snake_case")]
244#[non_exhaustive]
245pub enum E2eBasis {
246    /// Framework ingest time — immune to producer clock skew (default).
247    #[default]
248    Ingest,
249    /// Record event time (e.g. Kafka message timestamp) — measures true
250    /// pipeline lag but is sensitive to upstream clocks.
251    Event,
252}
253
254mod defaults {
255    pub(super) fn io_threads() -> usize {
256        2
257    }
258}
259
260impl PipelineConfig {
261    /// Load from YAML text: interpolate `${VAR}` forms against the process
262    /// environment, parse, and validate.
263    // An inherent `from_str` (rather than `std::str::FromStr`) keeps the
264    // call site `PipelineConfig::from_str(text)?` working without a trait
265    // import, matching `from_path` beside it.
266    #[expect(
267        clippy::should_implement_trait,
268        reason = "paired with from_path; no trait import required at call sites"
269    )]
270    pub fn from_str(text: &str) -> Result<Self, ConfigError> {
271        let interpolated = interpolate::interpolate(text)?;
272        Self::parse_interpolated(&interpolated)
273    }
274
275    /// Load from a YAML file (read, interpolate, parse, validate).
276    pub fn from_path(path: &Path) -> Result<Self, ConfigError> {
277        let text = std::fs::read_to_string(path).map_err(|source| ConfigError::Io {
278            path: path.to_owned(),
279            source,
280        })?;
281        Self::from_str(&text)
282    }
283
284    fn parse_interpolated(text: &str) -> Result<Self, ConfigError> {
285        let de = serde_yaml::Deserializer::from_str(text);
286        let mut cfg: PipelineConfig =
287            serde_path_to_error::deserialize(de).map_err(|e| ConfigError::Parse {
288                path: e.path().to_string(),
289                source: e.into_inner(),
290            })?;
291        cfg.source.set_section("source");
292        cfg.sink.set_section("sink");
293        if let Some(deser) = cfg.deserializer.as_mut() {
294            deser.set_section("deserializer");
295        }
296        cfg.validate()?;
297        Ok(cfg)
298    }
299
300    /// Cross-field validation, run automatically by the loaders. Public so
301    /// programmatically built configs (tests, `etl-test`) get the same
302    /// checks.
303    pub fn validate(&self) -> Result<(), ConfigError> {
304        let fail = |msg: String| Err(ConfigError::Validation(msg));
305
306        if self.pipeline.name.trim().is_empty() {
307            return fail("pipeline.name must not be empty".into());
308        }
309        if self.pipeline.io_threads == 0 {
310            return fail("pipeline.io_threads must be at least 1".into());
311        }
312        if self.pipeline.threads == Some(0) {
313            return fail("pipeline.threads must be at least 1 when set".into());
314        }
315        // The commit loop fires whenever `last_commit.elapsed() >= interval`,
316        // so an interval below a poll cycle commits on nearly every loop.
317        // A floor keeps sub-100ms intervals from hammering the source's
318        // offset store for no durability gain (checkpointing is not the
319        // durability boundary — the sink write is).
320        const MIN_COMMIT_INTERVAL: Duration = Duration::from_millis(100);
321        if self.checkpoint.interval < MIN_COMMIT_INTERVAL {
322            return fail(format!(
323                "checkpoint.interval must be at least 100ms (got {:?}): the commit \
324                 loop fires every interval, so sub-100ms intervals hammer the \
325                 source's offset store without improving durability",
326                self.checkpoint.interval
327            ));
328        }
329        if self.checkpoint.max_pending_batches == 0 {
330            return fail("checkpoint.max_pending_batches must be at least 1".into());
331        }
332        if self.checkpoint.drain_timeout.is_zero() {
333            return fail("checkpoint.drain_timeout must be greater than zero".into());
334        }
335        if self.checkpoint.stalled_fail_after.is_zero() {
336            return fail("checkpoint.stalled_fail_after must be greater than zero".into());
337        }
338        if self.backpressure.max_inflight_bytes.as_u64() == 0 {
339            return fail("backpressure.max_inflight_bytes must be greater than zero".into());
340        }
341        let (low, high) = (self.backpressure.low_ratio, self.backpressure.high_ratio);
342        if !(low > 0.0 && low < high && high <= 1.0) {
343            return fail(format!(
344                "backpressure ratios must satisfy 0 < low_ratio < high_ratio <= 1 \
345                 (got low_ratio={low}, high_ratio={high})"
346            ));
347        }
348        Ok(())
349    }
350}
351
352#[cfg(test)]
353mod tests {
354    use super::*;
355
356    const MINIMAL: &str = r#"
357pipeline: { name: demo }
358source: { memory: {} }
359sink: { memory: {} }
360"#;
361
362    #[test]
363    fn minimal_config_applies_documented_defaults() {
364        let cfg = PipelineConfig::from_str(MINIMAL).unwrap();
365        assert_eq!(cfg.pipeline.name, "demo");
366        assert_eq!(cfg.pipeline.threads, None);
367        assert_eq!(cfg.pipeline.io_threads, 2);
368        assert_eq!(cfg.pipeline.pinning, PinningMode::Off);
369        assert_eq!(cfg.checkpoint.interval, Duration::from_secs(5));
370        assert_eq!(cfg.checkpoint.max_pending_batches, 1024);
371        assert_eq!(cfg.checkpoint.drain_timeout, Duration::from_secs(25));
372        assert_eq!(cfg.checkpoint.stalled_fail_after, Duration::from_secs(120));
373        assert_eq!(cfg.backpressure.max_inflight_bytes, ByteSize::mib(256));
374        assert_eq!(cfg.backpressure.high_ratio, 0.8);
375        assert_eq!(cfg.backpressure.low_ratio, 0.5);
376        assert_eq!(cfg.backpressure.min_pause, Duration::from_millis(500));
377        assert_eq!(cfg.metrics.exporter, MetricsExporter::Prometheus);
378        assert_eq!(cfg.metrics.listen, SocketAddr::from(([0, 0, 0, 0], 9090)));
379        assert!(!cfg.metrics.per_partition_detail);
380        assert_eq!(cfg.metrics.e2e_basis, E2eBasis::Ingest);
381        assert!(cfg.deserializer.is_none());
382    }
383
384    #[test]
385    fn full_design_doc_example_parses() {
386        // The connector bodies below mirror the module-doc example and use the
387        // real connector field names. etl-core has no dependency on the
388        // connector crates, so this test can only parse the framework layer;
389        // the connector body shapes are verified by hand against
390        // crates/etl-kafka/src/config.rs (brokers/topic/group_id),
391        // crates/etl-avro/src/config.rs (registry.url), and
392        // crates/etl-clickhouse/src/config.rs (table/columns/shards) — and
393        // stay consistent with crates/etl/examples/kafka_avro_to_clickhouse.yaml.
394        let yaml = r#"
395pipeline: { name: orders, threads: 4, io_threads: 2 }
396checkpoint: { interval: 5s, max_pending_batches: 1024 }
397backpressure: { max_inflight_bytes: 256MiB }
398source:
399  kafka:
400    brokers: ${KAFKA_BROKERS:-localhost:9092}
401    topic: orders
402    group_id: orders-etl
403    rdkafka: { fetch.message.max.bytes: "1048576" }
404deserializer:
405  avro:
406    mode: confluent
407    registry:
408      url: "${SCHEMA_REGISTRY_URL:-http://sr:8081}"
409sink:
410  clickhouse:
411    table: orders_local
412    columns: [id, amount, ts]
413    shards:
414      - { replicas: ["http://ch-0-0:8123", "http://ch-0-1:8123"] }
415      - { replicas: ["http://ch-1-0:8123", "http://ch-1-1:8123"] }
416    batch: { max_rows: 500000, max_bytes: 128MiB, linger: 1s }
417    inflight: { max_per_shard: 2 }
418    retry: { initial: 100ms, max: 10s, multiplier: 2.0 }
419metrics: { exporter: prometheus, listen: 0.0.0.0:9090 }
420"#;
421        let cfg = PipelineConfig::from_str(yaml).unwrap();
422        assert_eq!(cfg.pipeline.threads, Some(4));
423        assert_eq!(cfg.source.type_tag(), "kafka");
424        assert_eq!(cfg.deserializer.as_ref().unwrap().type_tag(), "avro");
425        assert_eq!(cfg.sink.type_tag(), "clickhouse");
426
427        // Interpolated default landed inside the opaque body, and the kafka
428        // body carries the required group_id.
429        #[derive(Debug, serde::Deserialize)]
430        struct KafkaProbe {
431            brokers: String,
432            group_id: String,
433            #[serde(flatten)]
434            _rest: serde_yaml::Value,
435        }
436        let kafka: KafkaProbe = cfg.source.deserialize_into().unwrap();
437        assert_eq!(kafka.brokers, "localhost:9092");
438        assert_eq!(kafka.group_id, "orders-etl");
439
440        // The avro body uses the nested `registry.url` shape, and the
441        // clickhouse body carries the required `columns`.
442        #[derive(Debug, serde::Deserialize)]
443        struct AvroProbe {
444            registry: RegistryProbe,
445        }
446        #[derive(Debug, serde::Deserialize)]
447        struct RegistryProbe {
448            url: String,
449        }
450        let avro: AvroProbe = cfg
451            .deserializer
452            .as_ref()
453            .unwrap()
454            .deserialize_into()
455            .unwrap();
456        assert_eq!(avro.registry.url, "http://sr:8081");
457
458        #[derive(Debug, serde::Deserialize)]
459        struct ChProbe {
460            columns: Vec<String>,
461        }
462        let ch: ChProbe = cfg.sink.deserialize_into().unwrap();
463        assert_eq!(ch.columns, ["id", "amount", "ts"]);
464    }
465
466    #[test]
467    fn unknown_fields_are_rejected_at_every_typed_level() {
468        for (yaml, field) in [
469            (
470                "pipeline: { name: x, bogus: 1 }\nsource: { m: {} }\nsink: { m: {} }",
471                "bogus",
472            ),
473            (
474                "pipeline: { name: x }\ncheckpoint: { intervall: 5s }\nsource: { m: {} }\nsink: { m: {} }",
475                "intervall",
476            ),
477            (
478                "pipeline: { name: x }\nmetrics: { port: 9 }\nsource: { m: {} }\nsink: { m: {} }",
479                "port",
480            ),
481            (
482                "pipeline: { name: x }\nsource: { m: {} }\nsink: { m: {} }\nsinks: {}",
483                "sinks",
484            ),
485        ] {
486            let err = PipelineConfig::from_str(yaml).unwrap_err().to_string();
487            assert!(err.contains(field), "expected `{field}` in error: {err}");
488        }
489    }
490
491    #[test]
492    fn parse_errors_carry_the_yaml_path() {
493        let yaml = "pipeline: { name: x, io_threads: many }\nsource: { m: {} }\nsink: { m: {} }";
494        let err = PipelineConfig::from_str(yaml).unwrap_err();
495        let text = err.to_string();
496        assert!(text.contains("pipeline.io_threads"), "{text}");
497    }
498
499    #[test]
500    fn validation_rules() {
501        let cases = [
502            (
503                "pipeline: { name: '  ' }\nsource: { m: {} }\nsink: { m: {} }",
504                "pipeline.name",
505            ),
506            (
507                "pipeline: { name: x, io_threads: 0 }\nsource: { m: {} }\nsink: { m: {} }",
508                "io_threads",
509            ),
510            (
511                "pipeline: { name: x, threads: 0 }\nsource: { m: {} }\nsink: { m: {} }",
512                "threads",
513            ),
514            (
515                "pipeline: { name: x }\ncheckpoint: { interval: 0s }\nsource: { m: {} }\nsink: { m: {} }",
516                "interval",
517            ),
518            (
519                "pipeline: { name: x }\ncheckpoint: { interval: 50ms }\nsource: { m: {} }\nsink: { m: {} }",
520                "at least 100ms",
521            ),
522            (
523                "pipeline: { name: x }\ncheckpoint: { max_pending_batches: 0 }\nsource: { m: {} }\nsink: { m: {} }",
524                "max_pending_batches",
525            ),
526            (
527                "pipeline: { name: x }\ncheckpoint: { drain_timeout: 0s }\nsource: { m: {} }\nsink: { m: {} }",
528                "drain_timeout",
529            ),
530            (
531                "pipeline: { name: x }\ncheckpoint: { stalled_fail_after: 0s }\nsource: { m: {} }\nsink: { m: {} }",
532                "stalled_fail_after",
533            ),
534            (
535                "pipeline: { name: x }\nbackpressure: { max_inflight_bytes: 0 }\nsource: { m: {} }\nsink: { m: {} }",
536                "max_inflight_bytes",
537            ),
538            (
539                "pipeline: { name: x }\nbackpressure: { low_ratio: 0.9, high_ratio: 0.8 }\nsource: { m: {} }\nsink: { m: {} }",
540                "low_ratio",
541            ),
542            (
543                "pipeline: { name: x }\nbackpressure: { high_ratio: 1.5 }\nsource: { m: {} }\nsink: { m: {} }",
544                "high_ratio",
545            ),
546        ];
547        for (yaml, needle) in cases {
548            let err = PipelineConfig::from_str(yaml).unwrap_err().to_string();
549            assert!(err.contains(needle), "expected `{needle}` in: {err}");
550        }
551    }
552
553    #[test]
554    fn missing_required_sections_error_clearly() {
555        let err = PipelineConfig::from_str("pipeline: { name: x }\nsink: { m: {} }")
556            .unwrap_err()
557            .to_string();
558        assert!(err.contains("source"), "{err}");
559        let err = PipelineConfig::from_str("source: { m: {} }\nsink: { m: {} }")
560            .unwrap_err()
561            .to_string();
562        assert!(err.contains("pipeline"), "{err}");
563    }
564
565    #[test]
566    fn interpolation_failures_surface_with_position() {
567        let err = PipelineConfig::from_str("pipeline:\n  name: ${UNSET_VAR_FOR_TEST}\n")
568            .unwrap_err()
569            .to_string();
570        assert!(err.contains("UNSET_VAR_FOR_TEST"), "{err}");
571        assert!(err.contains("line 2"), "{err}");
572    }
573
574    #[test]
575    fn from_path_reads_interpolates_and_reports_io_errors() {
576        use std::io::Write as _;
577        let mut file = tempfile::NamedTempFile::new().unwrap();
578        write!(
579            file,
580            "pipeline: {{ name: ${{FILE_TEST_NAME:-from-file}} }}\nsource: {{ m: {{}} }}\nsink: {{ m: {{}} }}\n"
581        )
582        .unwrap();
583        let cfg = PipelineConfig::from_path(file.path()).unwrap();
584        assert_eq!(cfg.pipeline.name, "from-file");
585
586        let err = PipelineConfig::from_path(Path::new("/nonexistent/etl.yaml")).unwrap_err();
587        assert!(matches!(err, ConfigError::Io { .. }));
588        assert!(err.to_string().contains("/nonexistent/etl.yaml"));
589    }
590
591    #[test]
592    fn equal_inputs_parse_to_equal_configs() {
593        let a = PipelineConfig::from_str(MINIMAL).unwrap();
594        let b = PipelineConfig::from_str(MINIMAL).unwrap();
595        assert_eq!(a, b);
596        let c = PipelineConfig::from_str(
597            "pipeline: { name: other }\nsource: { m: {} }\nsink: { m: {} }",
598        )
599        .unwrap();
600        assert_ne!(a, c);
601    }
602}
etl_core/config/mod.rs

etl_core/config/
mod.rs
