[mwarkentin]

Problem

test_old_queue_is_empty_when_the_swap_happens fails now and then with left: 1, right: 0.

The test reads in_flight_count() on the old client at the swap. That is librdkafka's rd_kafka_outq_len. It counts queued messages, but also stats reports, logs and delivery reports that are waiting to be served by poll.

Arroyo's producer config sends a stats report every second. lingering_blob sets linger.ms to 1000, so the drain usually ends about 1 s after the client is built, right when its first stats report comes in. If that report is still waiting on the queue when the test reads the count, the count is 1 and the assert fails.

The same miscount hits production code. After the drain, swap() used old.in_flight_count() as the number of purged messages. A waiting stats report then logs "dropping the old client with messages still queued, they are lost" and bumps arroyo.producer.config_reload_purged_messages, though nothing was lost. test_ignoring_ordering_still_drains_the_retired_client checks that count, so it can flake the same way.

Evidence:

• CI failed 4 times on this since the tests landed in Add ReloadingKafkaProducer #573, always with left: 1, right: 0:

  • test_ignoring_ordering_still_drains_the_retired_client, rdkafka 0.37, on main at 49c24ec (job)
  • test_old_queue_is_empty_when_the_swap_happens, rdkafka 0.38, 8cb627c (job)
  • test_old_queue_is_empty_when_the_swap_happens, rdkafka 0.37, f99b937 (job)
  • test_old_queue_is_empty_when_the_swap_happens, rdkafka 0.37, first run of fix(rust): Remove duplicate imports in producer tests #599 (job)

  The other red Rust runs in that time were the build errors fixed by fix(rust): Remove duplicate imports in producer tests #599 and one job that hit the 10 minute limit.

• I timed it on a copy of main with extra logging. With rdkafka 0.37, the old client's first stats report was served 1001.3 to 1002.2 ms after the client was built, and the drain ended at 1001.8 to 1002.7 ms (15 of 15 runs). With 0.39 it was the same in 8 of 10 runs. In the other 2, the probe took about 50 ms longer, and so did the drain.

• I could not make the unmodified test fail on this machine: 0 failures in 530 runs on main (rdkafka 0.37 and 0.39, idle, next to busy loops, pinned to one CPU, and next to other copies of the tests). It is a narrow race, and CI runners are busier.

• On a copy of main with a shorter stats interval in lingering_blob (rdkafka 0.39), it fails the same way, with left: 1. That was 1 of 30 runs at 1 ms, 3 of 130 at 5 ms, and 0 of 30 at 20 ms. With stats off it was 0 of 100. With rdkafka 0.37 it did not fail even at 1 ms.

Change

1. Test fix: lingering_blob turns stats off ("statistics.interval.ms": "0"). Only the two lingering tests use it. Both assert on messages, not stats.

2. Production fix, in its own commit: drain() now returns whether the flush finished. If it did, swap() reports 0 purged messages. If the drain timed out, swap() still uses in_flight_count().

   This is exact after a drain that finished. rd_kafka_flush only succeeds once every message is delivered or failed and its delivery report has been served. After that, nothing can reach the old client. After a timed-out drain the count is unchanged. It is an upper bound: it can count too many, never too few.

3. Two new tests:

   • test_stats_waiting_on_a_drained_client_are_not_counted_as_purged holds the old client's polling thread in a stats callback, so a report waits on its queue during the swap.
   • test_timed_out_drain_counts_the_messages_it_leaves checks that a timed-out drain still counts the message it leaves behind.

   purged_at_drop (test only) is now an Option, so a test can read it before the old client's drop finishes.

How tested

Before, on main: see the numbers above.

After, on this branch:

• test_old_queue_is_empty_when_the_swap_happens with cargo test --lib -- --exact: 150 of 150 passed in a row on rdkafka 0.39, 150 of 150 on 0.37, and 100 of 100 on 0.38.
• The whole reloading_producer module (30 tests): 5 runs each on rdkafka 0.37, 0.38 and 0.39, all passed. It also passed once with --features pull.
• The lingering tests still catch what they are for. With the drain before the install removed, test_old_queue_is_empty_when_the_swap_happens fails 5 of 5. With the drain after the install removed (unordered path), test_ignoring_ordering_still_drains_the_retired_client fails 5 of 5.
• The purge tests:
  • With the old purge count, test_stats_waiting_on_a_drained_client_are_not_counted_as_purged fails every time with left: 1: 20 of 20 runs on rdkafka 0.38 and 25 of 25 on 0.39. On 0.37 it passes either way. There flush keeps polling until in_flight_count() is 0, stats reports included, so the report is served before the count. From 0.38 on, flush returns as soon as no messages are left.
  • With the fix, both new tests pass 20 of 20 on each version.
  • With the purge count forced to 0, test_timed_out_drain_counts_the_messages_it_leaves fails 5 of 5.
• cargo fmt --check, cargo check and cargo check --features pull pass. CI does not run clippy. cargo clippy --all-targets --features pull shows the same warnings as on main, and none in reloading_producer.rs.

Follow-up

• After a timed-out drain, the purge count can still include waiting stats reports, logs or delivery reports. Getting it exact means counting the delivery reports that fail with a purge error when the old client is dropped. Those go to the caller's context, so it would need a wrapper context. That is too big for this PR.
• test_produce_does_not_wait_for_a_drain_that_never_finishes also reads in_flight_count() with stats on. It has not flaked in CI, so I left it alone.

---

Generated by Claude Code

[linear-code]

STREAM-2030