queue: honour poll timeout between handled internal ops (#5325)

[Tamir Suliman (allamiro)]

Fixes #5325.

Problem

rd_kafka_poll(0) / rd_kafka_consumer_poll(0) (and other queue poll variants) can block far longer than the requested timeout — up to indefinitely — when internal ops are enqueued faster than the application thread drains them. The reported trigger is debug=all + log.queue=true (which routes log messages as RD_KAFKA_OP_LOG onto the polled queue), where broker threads flood the queue with log ops. The issue reporter observed consumer_poll(0) blocking for over 2 hours in production.

Root cause

In rd_kafka_q_pop_serve0() the timeout is only evaluated once the queue is found empty (at the cnd_timedwait_abs() path). After handling an internal op (RD_KAFKA_OP_RES_HANDLED / RD_KAFKA_OP_RES_KEEP, e.g. a log op) the loop does goto retry without re-checking the timeout. If internal ops keep arriving the queue never empties, so the timeout path is never reached and the call is starved.

Fix

Re-check the timeout after each handled op, before jumping back to retry. At least one op is always handled per call, so forward progress is guaranteed, while control is returned to the application once the timeout has expired.

This matches the fix proposed and verified in the issue, with one addition: the early return also calls rd_kafka_app_polled() when the queue can contain fetched messages, consistent with every other return path after rd_kafka_app_poll_start(). Omitting it (as in the issue snippet) would leave the max.poll.interval.ms bookkeeping unbalanced.

Behaviour note for reviewers

For an already-expired timeout (i.e. poll(0)), the call now handles one internal op and returns, rather than draining all currently-queued internal ops in a single call. This is bounded and FIFO-safe (a returnable message behind buffered internal ops surfaces on a subsequent call; it is never lost or reordered), and is the same semantics as the verified fix in the issue. Polls with a non-zero timeout drain until the timeout elapses, as before.

Test

Adds a deterministic regression unit test q_pop_serve_starvation (src/rdkafka_queue.c, registered in src/rdunittest.c) that continuously feeds the queue during an RD_POLL_NOWAIT poll and asserts the poll honours its timeout (handles exactly one op).

• Fails on unpatched code (drains all fed ops in a single NOWAIT poll).
• Passes with the fix.
• Full unit-test suite (25 tests) passes.
• make style-check-changed clean (clang-format 18).

A CHANGELOG.md entry is included under Fixes.

[confluent-cla-assistant]

🎉 All Contributor License Agreements have been signed. Ready to merge.
✅ allamiro
_{Please push an empty commit if you would like to re-run the checks to verify CLA status for all contributors.}