From ee33d4dc5855374cb7e2bd612eaa74db10b25d85 Mon Sep 17 00:00:00 2001 From: Leonid Ryzhyk Date: Fri, 3 Jul 2026 02:31:26 -0400 Subject: [PATCH] [adapters] Tolerate transient Kafka errors in FT input tests Transient librdkafka connectivity blips ("AllBrokersDown", "N/M brokers are down") are reported to the consumer as non-fatal errors, and the connector recovers on its own. DummyInputConsumer recorded every such error in the strict call sequence that expect() checks, so a blip anywhere in a test failed it, e.g.: thread 'transport::kafka::ft::test::multiple_input' panicked at crates/adapters/src/transport/kafka/ft/test.rs:463:22: assertion `left == right` failed left: [Extended { num_records: 90, ... }] right: [Error(false)] Record only fatal errors. No test expects ConsumerCall::Error, so no expectation changes. Signed-off-by: Leonid Ryzhyk --- crates/adapters/src/transport/kafka/ft/test.rs | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/crates/adapters/src/transport/kafka/ft/test.rs b/crates/adapters/src/transport/kafka/ft/test.rs index d32e7c56ce7..cb2425a83bd 100644 --- a/crates/adapters/src/transport/kafka/ft/test.rs +++ b/crates/adapters/src/transport/kafka/ft/test.rs @@ -757,7 +757,12 @@ impl InputConsumer for DummyInputConsumer { fn error(&self, fatal: bool, error: AnyError, _tag: Option<&'static str>) { info!("error: {error}"); - self.called(ConsumerCall::Error(fatal)); + // Record only fatal errors. Non-fatal errors, such as a momentary + // broker disconnection, can arrive at any time, and recording them + // would break the strict call sequence that `expect()` checks. + if fatal { + self.called(ConsumerCall::Error(fatal)); + } } fn request_step(&self) {}