fix(ci): prevent kafka consumer teardown hang with xdist

[mchen-sentry]

Observed these 99% completion hang failures after xdist. Reproduced on a test branch and a thread dump on the hung process showed:

File "arroyo/backends/kafka/consumer.py", line 348 in assignment_callback
File "arroyo/backends/kafka/consumer.py", line 719 in close
File "arroyo/processing/processor.py", line 567 in _shutdown
File "arroyo/processing/processor.py", line 353 in run
File "sentry/testutils/pytest/kafka.py", line 87 in scope_consumers

Let's run consumer.run() on a daemon thread with a 5-second join timeout. This still attempts a graceful shutdown, but bails out if the broker is slow. If the thread outlives the join, it dies on process exit and the broker detects the dead consumer via heartbeat timeout.

Tested that hangs no longer happen with the fix.

We could also potentially just remove consumer.run() entirely and skip graceful shutdown as the fixture is session-scoped so teardown runs right before process exit.

[linear-code]

DI-1753 fix: kafka ci shard hang

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

Autofix Details

Bugbot Autofix prepared a fix for the issue found in the latest run.

• ✅ Fixed: Missing try/except lets one consumer failure skip others
  • Added try/except block around signal_shutdown() call to ensure one consumer's failure doesn't prevent cleanup of remaining consumers in the loop.

Or push these changes by commenting:

@cursor push bf4c8e5e4f

Preview (bf4c8e5e4f)

diff --git a/src/sentry/testutils/pytest/kafka.py b/src/sentry/testutils/pytest/kafka.py
--- a/src/sentry/testutils/pytest/kafka.py
+++ b/src/sentry/testutils/pytest/kafka.py
@@ -81,7 +81,10 @@
 
     for consumer_name, consumer in all_consumers.items():
         if consumer is not None:
-            consumer.signal_shutdown()
+            try:
+                consumer.signal_shutdown()
+            except Exception:
+                _log.warning("Could not shutdown consumer %s", consumer_name)
 
 
 @pytest.fixture(scope="function")

_{This Bugbot Autofix run was free. To enable autofix for future PRs, go to the Cursor dashboard.}

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

^{Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.}