Cover split partition queues with the periodic fallback wakeup

[davidak09]

Summary

StreamConsumer::split_partition_queue can permanently stall a single partition's stream after a lost wakeup, while every other partition keeps consuming. This adds the periodic fallback wakeup to split partition queues that the main consumer queue already has.

Background

A parked MessageStream is rewoken by an edge-triggered "queue non-empty" callback (native_message_queue_nonempty_cb) that only fires on an empty → non-empty transition. That edge can be missed (see #665), leaving the stream parked with no pending wakeup.

The main consumer queue is protected against this: from_config_and_context spawns a background loop that calls wakers.wake_all() every max.poll.interval.ms / 2, so a missed edge self-heals within at most that interval.

split_partition_queue creates its own WakerSlab and enables its own non-empty callback — but no equivalent fallback loop is spawned for it. So for split partition queues a missed wakeup is not recovered: the stream stays parked indefinitely.

Impact

Consumers that use split_partition_queue (one stream per partition for parallel/independent draining) can have a single partition silently stop consuming for hours/days while sibling partitions on the same consumer keep flowing. Only recreating the consumer clears it. This matches behavior reported downstream in vectordotdev/vector#22006 (one partition's lag climbs unbounded; the consumer is otherwise healthy, no errors logged).

Fix

When split_partition_queue creates the queue's WakerSlab, spawn the same fallback loop used for the main queue, waking that queue's wakers every max.poll.interval.ms / 2. The loop holds a Weak<WakerSlab> and exits once the queue is dropped and its WakerSlab is freed, so no shutdown plumbing or changes to StreamPartitionQueue are needed. poll_interval is now stored on StreamConsumer to reuse the value parsed at construction.

Net production change: one file, +10/−16.

Test

tests/test_partition_queue_wakeup.rs — a deterministic, broker-free regression test. It assigns and splits a partition, parks the stream by polling it with a custom waker (empty queue, no broker → no data), and asserts the stream is woken within the timeout. The only thing that can wake it is the fallback loop.

Verified fail → pass: with the fallback loop removed the test times out (never woken); with the fix it passes in ~0.5s.

Validation

• cargo test --lib — passes (no broker).
• New regression test — passes with the fix, fails without it.
• Existing consumer integration tests (incl. the other split_partition_queue tests) — pass against a real broker (docker compose up + cargo test), no regression.

Notes

Relates to #665; addresses the stuck-partition symptom in vectordotdev/vector#22006.

Developed with AI assistance and reviewed by me.