{"record":{"id":"9a45e9a3369df53f","repo":"vectordotdev/vector","slug":"the-drain-timeout-ms-must-be-less-than-sessio","errorCode":null,"errorMessage":"The drain_timeout_ms ({}) must be less than session_timeout_ms ({})","messagePattern":"The drain_timeout_ms \\((.+?)\\) must be less than session_timeout_ms \\((.+?)\\)","errorType":"validation","errorClass":"BuildError","httpStatus":null,"severity":"error","filePath":"src/sources/kafka.rs","lineNumber":347,"sourceCode":"\n#[async_trait::async_trait]\n#[typetag::serde(name = \"kafka\")]\nimpl SourceConfig for KafkaSourceConfig {\n    async fn build(&self, cx: SourceContext) -> crate::Result<super::Source> {\n        let log_namespace = cx.log_namespace(self.log_namespace);\n\n        let decoder =\n            DecodingConfig::new(self.framing.clone(), self.decoding.clone(), log_namespace)\n                .build()?;\n        let decompressor = self\n            .decompression\n            .as_ref()\n            .map(DecompressionConfig::build)\n            .transpose()?;\n        let acknowledgements = cx.do_acknowledgements(self.acknowledgements);\n\n        if let Some(d) = self.drain_timeout_ms {\n            snafu::ensure!(\n                Duration::from_millis(d) <= self.session_timeout_ms,\n                InvalidDrainTimeoutSnafu {\n                    value: d,\n                    session_timeout_ms: self.session_timeout_ms\n                }\n            );\n        }\n\n        let (consumer, callback_rx) = create_consumer(self, acknowledgements)?;\n\n        Ok(Box::pin(kafka_source(\n            self.clone(),\n            consumer,\n            callback_rx,\n            decoder,\n            decompressor,\n            cx.out,\n            cx.shutdown,","sourceCodeStart":329,"sourceCodeEnd":365,"githubUrl":"https://github.com/vectordotdev/vector/blob/3708c39b12a93212ed8b8d7510b4cc7769cb5864/src/sources/kafka.rs#L329-L365","documentation":"The Kafka source validates that an explicit `drain_timeout_ms` (how long the source waits to flush pending acknowledgements on shutdown) does not exceed `session_timeout_ms`. If it does, `InvalidDrainTimeoutSnafu` fails the source build: draining longer than the consumer session timeout would be pointless because the broker evicts the consumer from the group first. Note the check is `<=`, so equal values are allowed; the default drain timeout is half of `session_timeout_ms` (10s default → 5s).","triggerScenarios":"Setting `drain_timeout_ms` in a `kafka` source config to a value greater than `session_timeout_ms` (default 10000). Example: `session_timeout_ms: 6000` with `drain_timeout_ms: 15000` fails at config build time, before any broker connection.","commonSituations":"Operators raising drain timeout for graceful shutdown without touching session timeout; lowering `session_timeout_ms` for faster rebalancing while keeping an old large `drain_timeout_ms`; copying drain settings between sources with different session timeouts.","solutions":["Set `drain_timeout_ms` to a value ≤ `session_timeout_ms` (or just remove it — the default is half the session timeout).","If a longer drain is genuinely needed, raise `session_timeout_ms` first, then set `drain_timeout_ms` under it (mind broker `group.max.session.timeout.ms`).","Re-run `vector validate <config>` to confirm the constraint passes before deploy."],"exampleFix":"# before\nsources:\n  kafka_in:\n    type: kafka\n    session_timeout_ms: 6000\n    drain_timeout_ms: 15000\n\n# after\nsources:\n  kafka_in:\n    type: kafka\n    session_timeout_ms: 6000\n    drain_timeout_ms: 5000","handlingStrategy":"validation","validationCode":"# Pre-deploy check (or `vector validate`):\npython3 - <<'EOF'\nimport yaml, sys\ncfg = yaml.safe_load(open(\"vector.yaml\"))\nfor s in (cfg.get(\"sources\") or {}).values():\n    if s.get(\"type\") == \"kafka\":\n        d, sess = s.get(\"drain_timeout_ms\"), s.get(\"session_timeout_ms\", 10000)\n        if d is not None and d > sess:\n            sys.exit(f\"drain_timeout_ms {d} > session_timeout_ms {sess}\")\nEOF","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Omit drain_timeout_ms and accept the safe default (half of session_timeout_ms).","When tuning either knob, tune them as a pair; keep drain ≤ session.","Run `vector validate` in CI for every config change."],"tags":["kafka","config","validation","shutdown","vector"],"backgroundTag":"invalid-config-value","analyzedSha":"3708c39b12a93212ed8b8d7510b4cc7769cb5864","analyzedAt":"2026-08-20T07:02:18.786Z","contentChangedAt":null,"schemaVersion":2},"datasetVersion":"2026-09-14T05:17:10.506Z"}