From 6923169612df5d6a569291813f1bb39ba15d2775 Mon Sep 17 00:00:00 2001
From: Derek <derek@hyperi.io>
Date: Mon, 12 Oct 2026 19:45:52 +1100
Subject: [PATCH] test: log the error when a Kafka test loader exits

kafka_offset_floor and kafka_rebalance spawn the orchestrator and threw away what run() returned, so a loader that failed to start left a test waiting on a timeout with nothing in its output. They now log the error the way grpc_outage does.
---
 tests/integration/kafka_offset_floor.rs | 4 +++-
 tests/integration/kafka_rebalance.rs    | 8 ++++++--
 2 files changed, 9 insertions(+), 3 deletions(-)

diff --git a/tests/integration/kafka_offset_floor.rs b/tests/integration/kafka_offset_floor.rs
index 392adfa..d93d52f 100644
--- a/tests/integration/kafka_offset_floor.rs
+++ b/tests/integration/kafka_offset_floor.rs
@@ -280,7 +280,9 @@ fn start_loader(config: Config) -> (CancellationToken, JoinHandle<()>) {
     let mut orchestrator = Orchestrator::new(config);
     let shutdown = orchestrator.shutdown_token();
     let loader = tokio::spawn(async move {
-        let _ = orchestrator.run().await;
+        if let Err(error) = orchestrator.run().await {
+            tracing::error!(%error, "the loader stopped with an error");
+        }
     });
     (shutdown, loader)
 }
diff --git a/tests/integration/kafka_rebalance.rs b/tests/integration/kafka_rebalance.rs
index 3789538..a05ce0d 100644
--- a/tests/integration/kafka_rebalance.rs
+++ b/tests/integration/kafka_rebalance.rs
@@ -157,7 +157,9 @@ async fn a_partition_moved_mid_buffer_lands_each_row_once() {
     let first_guard = Arc::clone(first.memory_guard());
     let stop_first = first.shutdown_token();
     let first = tokio::spawn(async move {
-        let _ = first.run().await;
+        if let Err(error) = first.run().await {
+            tracing::error!(%error, "the first loader stopped with an error");
+        }
     });
     let deadline = tokio::time::Instant::now() + BUDGET;
     let mut buffered = 0.0;
@@ -183,7 +185,9 @@ async fn a_partition_moved_mid_buffer_lands_each_row_once() {
     let second_guard = Arc::clone(second.memory_guard());
     let stop_second = second.shutdown_token();
     let second = tokio::spawn(async move {
-        let _ = second.run().await;
+        if let Err(error) = second.run().await {
+            tracing::error!(%error, "the second loader stopped with an error");
+        }
     });
     let taken_over = settled_rows(&clickhouse, &table).await;