fix(sink): set the producer batching delay once

config.example.yaml

@@ -93,7 +93,7 @@ dfe_source: "my_source"
 #   batch:
 #     max_events: 10000                # Vector-level batch size (default 10K)
 #     # max_bytes: 8388608             # optional byte limit, at most 16 MiB on direct, where unset means 8 MiB
-#     timeout_secs: 1                  # flush timeout, at most 1 on direct
+#     timeout_secs: 1                  # flush timeout, at most 1 on direct. On bus the profile's linger.ms replaces it
 #   sasl:
 #     enabled: true
 #     mechanism: "scram_sha_512"

docs/config-schema.json

@@ -606,7 +606,7 @@
           "default": null
         },
         "timeout_secs": {
-          "description": "Max batch age (seconds) before flush. Default: 1, and at most 1 on the\ndirect transport.",
+          "description": "Max batch age (seconds) before flush. Default: 1, and at most 1 on the\ndirect transport. On the bus transport it applies only where no\nlibrdkafka layer sets `linger.ms`, which the production profile does.",
           "type": "integer",
           "format": "uint32",
           "minimum": 0,

docs/config-schema.yaml

@@ -582,7 +582,8 @@ $defs:
       timeout_secs:
         description: |-
           Max batch age (seconds) before flush. Default: 1, and at most 1 on the
-          direct transport.
+          direct transport. On the bus transport it applies only where no
+          librdkafka layer sets `linger.ms`, which the production profile does.
         type: integer
         format: uint32
         minimum: 0

docs/librdkafka.md

@@ -23,7 +23,7 @@
 | `message.timeout.ms`, `socket.timeout.ms` | `sink.message_timeout_ms` (0, no limit), `sink.socket_timeout_ms` (60000) |
 | `queue.buffering.max.ms`, `batch.num.messages`, `batch.size` | `sink.batch.timeout_secs`, `.max_events` and, where set, `.max_bytes` |
 
-`linger.ms` is librdkafka's other name for `queue.buffering.max.ms`. rdkafka applies a client's options from a hash map, so with the production profile's `linger.ms` beside Vector's `queue.buffering.max.ms`, which one a producer ends up with is not fixed.
+`linger.ms` is librdkafka's other name for `queue.buffering.max.ms`, and a client built from both takes whichever its hash map applied last. Vector always writes `sink.batch.timeout_secs` as `queue.buffering.max.ms`, so the supervisor moves a batching delay that a layer names into `batch.timeout_secs` and leaves both spellings out of `librdkafka_options`. A layer that names either spelling replaces the other spelling from every layer below it, and within one layer the spelling that sorts last wins. A layer therefore outranks `sink.batch.timeout_secs`. The production profile names `linger.ms`, so on the bus transport `timeout_secs` has no effect until a central profile drops that key, and the supervisor logs a warning when `timeout_secs` is not 1 and a layer overrides it.
 
 A few keys are derived from the big dials and written after the merge, so `librdkafka_options` cannot reach them:
 
@@ -48,9 +48,9 @@
 
 ## Example
 
-For a producer with no central file, `sink.compression: lz4` and `sink.librdkafka_options: {linger.ms: "20"}`, the supervisor writes `linger.ms=20`, `compression.type=lz4`, `queue.buffering.max.kbytes=262144`, `enable.idempotence=true`, `socket.nagle.disable=true`, `statistics.interval.ms=1000` and `message.max.bytes=16777216`, the last because no layer set it.
+For a producer with no central file, `sink.compression: lz4` and `sink.librdkafka_options: {linger.ms: "20"}`, the supervisor writes `batch.timeout_secs: 0.02` on the sink and, in `librdkafka_options`, `compression.type=lz4`, `queue.buffering.max.kbytes=262144`, `enable.idempotence=true`, `socket.nagle.disable=true`, `statistics.interval.ms=1000` and `message.max.bytes=16777216`, the last because no layer set it.
 
-Add a central `producer.production` naming only `linger.ms: "50"` and `socket.nagle.disable` and `statistics.interval.ms` drop out of `librdkafka_options`, because the central profile replaced the constants rather than merging over them. Vector still writes `statistics.interval.ms=1000` itself. `linger.ms` stays `20`: the per-instance option outranks the central file. Drop that option and it becomes `50`.
+Add a central `producer.production` naming only `linger.ms: "50"` and `socket.nagle.disable` and `statistics.interval.ms` drop out of `librdkafka_options`, because the central profile replaced the constants rather than merging over them. Vector still writes `statistics.interval.ms=1000` itself. The delay stays 20 ms: the per-instance option outranks the central file. Drop that option and it becomes 50 ms.
 
 `dfe-transform-vector --config <file> assemble <dir>` writes `90_sink.yaml` and `00_source.yaml`, so you can read the result for any config.
 
@@ -69,7 +69,7 @@
 
 | Producer setting | Value | librdkafka default | Justification |
 |---------|-------|--------------------|---------------|
-| `linger.ms` | `100` | `5` | Larger batches -- 100 ms of latency for higher throughput. See Layers for how it meets Vector's `queue.buffering.max.ms`. |
+| `linger.ms` | `100` | `5` | Larger batches -- 100 ms of latency for higher throughput. This is the delay the sink uses, in place of `sink.batch.timeout_secs`. See Layers. |
 | `compression.type` | `zstd` | `none` | Best compression ratio with acceptable CPU cost. `sink.compression` replaces it on every sink. |
 | `socket.nagle.disable` | `true` | `true` | Already librdkafka's default. The profile states it. |
 | `statistics.interval.ms` | `1000` | `0` | librdkafka metrics. |
@@ -119,7 +119,7 @@
 
 ## Source code references
 
 - Profile constants: scalo-rs `src/kafka_config.rs`
 - Central file loader: `src/config/kafka_defaults.rs` (`consumer_profile()`, `producer_profile()`, the `OnceLock` cache, `merge_layers()`)
 - Layer merge and derived keys: `src/config/generate.rs` (`build_source_librdkafka`, `build_sink_librdkafka`, `SERVICE_CONSUMER_OVERRIDES`, `SERVICE_PRODUCER_OVERRIDES`)
 - Config fields: `src/config/loader.rs` (`SourceConfig.librdkafka_options`, `SinkConfig.librdkafka_options`)

pipelines/filebeat/filebeat.yaml

@@ -6,8 +6,8 @@
 # License:   BUSL-1.1
 # Copyright: (c) 2026 HYPERI PTY LIMITED
 #
-# A pre-canned transform file, dropped into `transforms.dir` as it stands: the
-# wrapper writes it into Vector's config dir verbatim, so the top-level
+# A pre-canned transform file for `transforms.dir`. The wrapper parses it and
+# writes it into Vector's config dir with its wired `inputs`, so the top-level
 # `enrichment_tables` key reaches Vector and the VRL's timezone lookups resolve.
 # `inputs: [dfe_source]` binds it to whatever Kafka topic `dfe_source` derives,
 # and the wrapper wires the transform's output to `dfe_sink`.
@@ -17,10 +17,10 @@
 # path or points `path` at its own copy; startup validation names the path when
 # the file is not there.
 #
-# The VRL body is byte-identical to dfe-transform-vrl pipelines/filebeat/
-# filebeat.vrl @ e0b7087, which is where it is maintained. Refresh by copying
+# The VRL body is dfe-transform-vrl pipelines/filebeat/filebeat.vrl @ e0b7087,
+# bar two lines of its header comment, and that repo maintains it. Refresh by copying
 # that file again rather than editing the copy here, so the two transform apps
-# keep grading against the same corpus.
+# keep passing the same cases.
 
 enrichment_tables:
   timezones:

src/config/generate.rs

@@ -27,7 +27,8 @@
 use std::collections::{BTreeMap, HashMap};
 
 use scalo::transport::kafka::MESSAGE_MAX_BYTES;
-use serde_json::{Map, Value};
+use serde_json::{Map, Number, Value};
+use tracing::warn;
 
 /// A YAML mapping in the generated Vector config.
 type Mapping = Map<String, Value>;
@@ -323,6 +324,7 @@ fn direct_sink_component(
         build_batch_block(
             &sink.batch,
             Some(sink.batch.max_bytes.unwrap_or(DIRECT_BATCH_MAX_BYTES)),
+            sink.batch.timeout_secs.into(),
         ),
     );
 
@@ -367,10 +369,28 @@ fn kafka_sink_component(sink: &SinkConfig, inputs: &[String], acknowledged: bool
         Value::Number(sink.socket_timeout_ms.into()),
     );
 
-    // Vector-level batch
+    // librdkafka options — production defaults + user overrides
+    let mut rdkafka = build_sink_librdkafka(sink);
+
+    // Vector always writes `batch.timeout_secs` as `queue.buffering.max.ms`,
+    // which librdkafka reads as `linger.ms`. A layer that names the delay
+    // moves into that field, so the sink states it once.
+    let timeout_secs = match take_batching_delay(&mut rdkafka) {
+        Some(layered) => {
+            if sink.batch.timeout_secs != BatchConfig::default().timeout_secs {
+                warn!(
+                    sink_batch_timeout_secs = sink.batch.timeout_secs,
+                    layered_delay_secs = %layered,
+                    "sink.batch.timeout_secs is ignored: a librdkafka layer sets the batching delay"
+                );
+            }
+            layered
+        }
+        None => sink.batch.timeout_secs.into(),
+    };
     component.insert(
         key("batch"),
-        build_batch_block(&sink.batch, sink.batch.max_bytes),
+        build_batch_block(&sink.batch, sink.batch.max_bytes, timeout_secs),
     );
 
     // SASL
@@ -386,21 +406,35 @@ fn kafka_sink_component(sink: &SinkConfig, inputs: &[String], acknowledged: bool
     // Buffer
     component.insert(key("buffer"), build_buffer_block(&sink.buffer));
 
-    // librdkafka options — production defaults + user overrides
-    let rdkafka = build_sink_librdkafka(sink);
     component.insert(key("librdkafka_options"), Value::Object(rdkafka));
 
     component
 }
 
+/// Remove the batching delay from a sink's librdkafka options and return it in
+/// seconds, the unit of Vector's `batch.timeout_secs`.
+///
+/// `None`, with the options untouched, when no layer names it or the value is
+/// not a number of milliseconds. librdkafka refuses such a value at startup.
+fn take_batching_delay(rdkafka: &mut Mapping) -> Option<Number> {
+    let name = ["linger.ms", "queue.buffering.max.ms"]
+        .into_iter()
+        .find(|name| rdkafka.contains_key(*name))?;
+    let millis: f64 = rdkafka.get(name)?.as_str()?.trim().parse().ok()?;
+    if !millis.is_finite() || millis < 0.0 {
+        return None;
+    }
+    let seconds = Number::from_f64(millis / 1000.0)?;
+    rdkafka.remove("linger.ms");
+    rdkafka.remove("queue.buffering.max.ms");
+    Some(seconds)
+}
+
 /// A sink's Vector-level `batch` block, with `max_bytes` written when given.
-fn build_batch_block(batch: &BatchConfig, max_bytes: Option<u64>) -> Value {
+fn build_batch_block(batch: &BatchConfig, max_bytes: Option<u64>, timeout_secs: Number) -> Value {
     let mut block = Mapping::new();
     block.insert(key("max_events"), Value::Number(batch.max_events.into()));
-    block.insert(
-        key("timeout_secs"),
-        Value::Number(batch.timeout_secs.into()),
-    );
+    block.insert(key("timeout_secs"), Value::Number(timeout_secs));
     if let Some(max_bytes) = max_bytes {
         block.insert(key("max_bytes"), Value::Number(max_bytes.into()));
     }
@@ -1352,7 +1386,6 @@ mod tests {
         assert!(text.contains("codec: json"));
         assert!(text.contains("compression: zstd"));
         // Production librdkafka defaults (from scalo kafka_config)
-        assert!(text.contains("linger.ms: \"100\""));
         assert!(text.contains("compression.type: zstd"));
         assert!(text.contains("socket.nagle.disable: \"true\""));
         assert!(text.contains("statistics.interval.ms: \"1000\""));
@@ -1374,7 +1407,10 @@ mod tests {
         assert!(text.contains("socket_timeout_ms: 60000"));
         // Vector batch
         assert!(text.contains("max_events: 10000"));
-        assert!(text.contains("timeout_secs: 1"));
+        // The production profile's `linger.ms` is the batching delay, in
+        // Vector's field.
+        assert!(text.contains("timeout_secs: 0.1"));
+        assert!(!text.contains("linger.ms"));
     }
 
     #[test]
@@ -1430,7 +1466,65 @@ mod tests {
 
         assert!(text.contains("max_events: 5000"));
         assert!(text.contains("max_bytes: 8388608"));
-        assert!(text.contains("timeout_secs: 2"));
+        // The production profile's `linger.ms` replaces `timeout_secs` on the bus.
+        assert!(text.contains("timeout_secs: 0.1"));
+    }
+
+    /// A Kafka sink states its batching delay once, as `batch.timeout_secs`.
+    /// Vector writes that as `queue.buffering.max.ms`, which librdkafka reads
+    /// as `linger.ms`, so a second statement leaves the producer with whichever
+    /// one the client applied last.
+    #[test]
+    fn a_kafka_sink_states_its_batching_delay_once() {
+        let with_option = |k: &str, v: &str| SinkConfig {
+            librdkafka_options: [(k.to_string(), v.to_string())].into(),
+            ..Default::default()
+        };
+
+        for (case, sink, seconds) in [
+            ("profile only", SinkConfig::default(), 0.1),
+            ("user linger.ms", with_option("linger.ms", "20"), 0.02),
+            (
+                "user queue.buffering.max.ms",
+                with_option("queue.buffering.max.ms", "5"),
+                0.005,
+            ),
+        ] {
+            let component = kafka_sink_component(&sink, &["src".into()], true);
+            let options = component["librdkafka_options"].as_object().unwrap();
+            assert!(!options.contains_key("linger.ms"), "{case}");
+            assert!(!options.contains_key("queue.buffering.max.ms"), "{case}");
+            assert_eq!(component["batch"]["timeout_secs"], seconds, "{case}");
+        }
+    }
+
+    /// The delay leaves the librdkafka options for Vector's own field, and a
+    /// value that is not a number of milliseconds stays where librdkafka
+    /// refuses it.
+    #[test]
+    fn the_batching_delay_moves_into_the_batch_timeout() {
+        let options = |pairs: &[(&str, &str)]| -> Mapping {
+            pairs.iter().map(|(k, v)| (k.to_string(), val(v))).collect()
+        };
+
+        for (case, pairs, seconds, left) in [
+            ("none", &[("fetch.min.bytes", "1")][..], None, 1),
+            ("linger", &[("linger.ms", "100")][..], Some(0.1), 0),
+            (
+                "alias",
+                &[("queue.buffering.max.ms", "1000")][..],
+                Some(1.0),
+                0,
+            ),
+            ("zero", &[("linger.ms", "0")][..], Some(0.0), 0),
+            ("not a number", &[("linger.ms", "soon")][..], None, 1),
+            ("negative", &[("linger.ms", "-5")][..], None, 1),
+        ] {
+            let mut rdkafka = options(pairs);
+            let delay = take_batching_delay(&mut rdkafka).map(|n| n.as_f64().unwrap());
+            assert_eq!(delay, seconds, "{case}");
+            assert_eq!(rdkafka.len(), left, "{case}");
+        }
     }
 
     /// The stack's producer compression is zstd, so a sink that names none

src/config/kafka_defaults.rs

@@ -180,11 +180,39 @@ pub fn producer_profile(profile_name: &str) -> HashMap<String, String> {
     }
 }
 
+/// Settings librdkafka accepts under two spellings and stores as one. A client
+/// built from a map applies its entries in hash order, so a map holding both
+/// spellings ends up with whichever was applied last.
+const ALIASES: &[(&str, &str)] = &[("linger.ms", "queue.buffering.max.ms")];
+
+/// The other spelling of `key`, when librdkafka has one.
+fn alias_of(key: &str) -> Option<&'static str> {
+    ALIASES.iter().find_map(|(a, b)| {
+        if key == *a {
+            Some(*b)
+        } else if key == *b {
+            Some(*a)
+        } else {
+            None
+        }
+    })
+}
+
+/// Write `key`, dropping the other spelling of the same setting.
+fn insert_unaliased(map: &mut HashMap<String, String>, key: &str, value: &str) {
+    if let Some(twin) = alias_of(key) {
+        map.remove(twin);
+    }
+    map.insert(key.to_string(), value.to_string());
+}
+
 /// Merge a base profile with user overrides.
 ///
 /// Applied in order, so the last writer wins: the central config (or its scalo
 /// fallback) baseline, then service-specific overrides, then the user's
-/// `librdkafka_options` from config YAML.
+/// `librdkafka_options` from config YAML. A layer that names a setting under
+/// one spelling replaces the other spelling from every layer below it, and
+/// within one layer the spelling that sorts last wins.
 #[must_use]
 #[allow(
     clippy::implicit_hasher,
@@ -195,14 +223,22 @@ pub fn merge_layers(
     service_overrides: &[(&str, &str)],
     user_overrides: &HashMap<String, String>,
 ) -> HashMap<String, String> {
-    let mut merged = base.clone();
+    let mut merged = HashMap::with_capacity(base.len());
+
+    let mut base_entries: Vec<(&String, &String)> = base.iter().collect();
+    base_entries.sort();
+    for (k, v) in base_entries {
+        insert_unaliased(&mut merged, k, v);
+    }
 
     for (k, v) in service_overrides {
-        merged.insert((*k).to_string(), (*v).to_string());
+        insert_unaliased(&mut merged, k, v);
     }
 
-    for (k, v) in user_overrides {
-        merged.insert(k.clone(), v.clone());
+    let mut user_entries: Vec<(&String, &String)> = user_overrides.iter().collect();
+    user_entries.sort();
+    for (k, v) in user_entries {
+        insert_unaliased(&mut merged, k, v);
     }
 
     merged
@@ -275,6 +311,68 @@ mod tests {
         assert_eq!(merged["queue.buffering.max.kbytes"], "524288");
     }
 
+    /// The setting under its other spelling never survives a higher layer, so
+    /// the merged map names it exactly once.
+    #[test]
+    fn merge_layers_keeps_one_spelling_of_an_aliased_setting() {
+        type Pairs = &'static [(&'static str, &'static str)];
+        let map = |pairs: &[(&str, &str)]| -> HashMap<String, String> {
+            pairs
+                .iter()
+                .map(|(k, v)| ((*k).to_string(), (*v).to_string()))
+                .collect()
+        };
+        let cases: [(Pairs, Pairs, Pairs, &str, &str); 6] = [
+            // (base, service, user, surviving key, surviving value)
+            (&[("linger.ms", "100")], &[], &[], "linger.ms", "100"),
+            (
+                &[("linger.ms", "100")],
+                &[],
+                &[("queue.buffering.max.ms", "5")],
+                "queue.buffering.max.ms",
+                "5",
+            ),
+            (
+                &[("queue.buffering.max.ms", "5")],
+                &[],
+                &[("linger.ms", "20")],
+                "linger.ms",
+                "20",
+            ),
+            (
+                &[("linger.ms", "100")],
+                &[("queue.buffering.max.ms", "7")],
+                &[],
+                "queue.buffering.max.ms",
+                "7",
+            ),
+            (
+                &[("linger.ms", "100"), ("queue.buffering.max.ms", "5")],
+                &[],
+                &[],
+                "queue.buffering.max.ms",
+                "5",
+            ),
+            (
+                &[],
+                &[],
+                &[("linger.ms", "20"), ("queue.buffering.max.ms", "5")],
+                "queue.buffering.max.ms",
+                "5",
+            ),
+        ];
+
+        for (base, service, user, key, value) in cases {
+            let merged = merge_layers(&map(base), service, &map(user));
+            let spellings: Vec<&str> = ["linger.ms", "queue.buffering.max.ms"]
+                .into_iter()
+                .filter(|k| merged.contains_key(*k))
+                .collect();
+            assert_eq!(spellings, vec![key], "base {base:?} user {user:?}");
+            assert_eq!(merged[key], value, "base {base:?} user {user:?}");
+        }
+    }
+
     #[test]
     fn parse_profile_map_handles_types() {
         let mut mapping = serde_json::Map::new();