Repository navigation
feat(server)!: add bounded deferred (long) polling - #4629
Merged
Merged
Conversation
Codecov Report❌ Patch coverage is ❌ Your patch check has failed because the patch coverage (10.64%) is below the target coverage (50.00%). You can increase the patch coverage or adjust the target coverage.
Additional details and impacted files@@ Coverage Diff @@
## master #4629 +/- ##
=============================================
- Coverage 90.73% 70.27% -20.46%
- Complexity 2989 3102 +113
=============================================
Files 1459 1465 +6
Lines 348351 356185 +7834
Branches 292933 300420 +7487
=============================================
- Hits 316084 250323 -65761
- Misses 25619 99107 +73488
- Partials 6648 6755 +107
🚀 New features to boost your workflow:
|
spetz
reviewed
Oct 24, 2026
spetz
left a comment
spetz
left a comment
Contributor
There was a problem hiding this comment.
Reviewed at f6eb3c3, master merge included. The merge resolution holds: the dispatch table, the metadata frontier gate and the HTTP routes carry both features. No critical findings. The four inline comments are warnings: a consumer that stalls with no signal on capacity refusals, two disk read budget problems in the deferred pump, and a breaking SDK default change without !. Smaller items (per-reply hashing and header walks on the pump, the Java frame clamp, doc drift) can follow separately.
9 tasks done
1 of 2 tasks
hubcio
force-pushed
the
feat/deferred-polling-pr
branch
from
October 26, 2026 11:16
f6eb3c3 to
84c4122
Compare
hubcio
changed the title
spetz
reviewed
Oct 26, 2026
numinnex
reviewed
Oct 26, 2026
Idle consumers repeatedly poll quiet partitions. Park requests until data arrives or the wait expires, with bounded deadlines, reply sizes and capacity. Default Rust consumers to deferred polling with manual commits; expose bounded polling through Java TCP and HTTP. Route readiness waits across partition leaders, preserve cancellation order and refresh budgets on each attempt. Keep progress ordering separate from producers, bound shared client admission, and distinguish capacity pressure from routing failures. Preserve retained offset operations and report uncertain delivery without replaying progress polls. Tie auto-commit fetches to demand, preserve partition positions and document lost-reply uncertainty. Cover limits, oversized records, cancellation, retries, session changes and cross-leader delivery. BREAKING CHANGE: Rust consumers default to AutoCommit::Disabled instead of committing on poll, and the stream builder starts at PollingStrategy::next() instead of last(). An application that never stores offsets reads everything again after a restart, and a new stream-builder group starts at offset zero. The consumer builder's poll_interval methods are gone, and IggyConsumerConfig::new, IggyConsumerConfig::from_stream_topic and IggyStreamConfig::new take DeferredPollOptions instead of a polling interval.
Java and PHP tests still use outdated polling API arguments. Java's protocol parity table omits deferred commands, and Swift's error table omits the new deferred capacity refusal. Update the test calls and synchronize the protocol fixtures with the server's command and error codes.
The worker dropped TransientNotAccepted and DeferredPollCapacityExceeded, which reach it only after the driver retried for the whole request timeout. A server that kept refusing left next() pending with no error, log or metric. Deliver both like any other poll failure, as master did. A readiness wait that times out on a lost replica now reaches the application as TransientNotAccepted, where a master poll with auto-commit reported TransientNotCommitted.
A waiter that reserve_read refused went back without a wake, so only a commit, its deadline or maintenance (16 waiters per 10 ms tick) probed it again, and waiters whose reads kept coming back took freed slots first. A refused read reached its request deadline although its messages were on disk. Refused reads now queue by first refusal. Only the oldest one may take a freed slot, and the pump wakes it when the budget has room.
A waiter whose read stayed below min_count read again from its start offset on every commit. On the disk tier each read reserved max_read_bytes, so a few such waiters held the shard's read budget for their whole wait and other disk reads were refused. The waiter now keeps the messages and bytes of its last read and counts only the resident messages committed after it. A full read starts once that total can reach min_count or the byte limit, so early replies at the byte cap still happen. When the new messages already left the journal, the waiter reads in full as before.
A waiter below min_count counted the messages committed after its last read with a trial read of the journal. That worked only for offset polls whose new messages were still resident. Timestamp polls, and every poll on a persisted single-replica topic, where a batch leaves the journal as it commits, still read in full on each commit. With the default min_count of 1, every filled wake built and selected its plan twice. The waiter now bounds the next read by the offsets and batch bytes committed after its last read. Offsets are consecutive, so the bound holds for every polling kind and needs no read. Partitions count the batch bytes that their commits make visible.
Freed read room pushed the oldest refused read to the back of the ready queue. Every disk read probed before it was refused, since none was first, so the room stayed unused while up to a full registry of waiters went ahead, each building a disk plan first. Only one refused read woke per pass, and room left after a service turn waited for the next 10 ms tick. Freed room now goes to the oldest refused read before other wakes, the pump keeps turning while one can start, and a queued read waits for its turn without building another plan.
The raw replica request refused polls and consumer-offset reads, because a partition read needs the partition's primary and its incarnation, and a request to a chosen replica carries neither. Deferred polls, readiness waits and deferred-request cancels passed the check. The server answers no cancel, so a raw cancel only waited for its deadline. The Rust and Java SDKs now refuse these codes before anything is sent.
The raw Java request sent deferred polls, readiness waits and deferred-request cancels as they were. A deferred poll carried no route and no partition context, a wait did not go to the primary of each partition it names, and a cancel only waited for its deadline. The Rust raw API routes the first two and refuses the third. A raw deferred poll now takes the routed path of a typed one, with its deadline set by the request timeout in its payload. Limits that the server's decoder refuses fail before anything is sent. A raw wait fails with FEATURE_UNAVAILABLE, because this client has no readiness-wait route, and a raw cancel fails with INVALID_COMMAND, as in Rust. Neither is sent.
A sink commits on poll. When a reply ends short of the partition head, the loop holds its messages until more arrive or the batch is full. A stop signal in that window ended the loop and dropped the held batch. The task then dropped the consumer without IggyConsumer::shutdown, so the batches inside the consumer stayed committed as well. The next run started after messages that the plugin never got. On a stop, or when the stream ends, the loop now gives the held batch to the plugin before it exits. The task then awaits IggyConsumer::shutdown, which moves the committed offset back to the last message that the plugin got and leaves the consumer group.
A readiness wait (command 107) takes no partitions after it starts. The worker kept one wait per group and skipped a group while any partition of it waited. A partition that drained in that time had no wait until that wait ended, which took up to max_wait. A committing member started waits only when no partition was ready. A backlog on one partition then hid new messages on the others until it drained. A partition without a known message and without a wait now restarts the wait of its group. Committing members keep their waits running while they fetch and between the polls of the application. A restart cancels the old wait, so each group still holds one deferred request. A second wait for each drained partition would hold more, and fetches wait behind them. New readiness groups restart every wait, so a restart never leaves a partition of another group behind a cancelled wait. A late reply of a replaced wait changes nothing.
A rebalance that left a partition with its member cleared its fetch position, so the partition restarted from its initial strategy. Under manual commits it read delivered messages again or, under Last, skipped some. Under commit-on-poll, a lost reply was retried with Next, which starts after the batch if the server committed it, so that batch never reached the application. The server keeps the owner of a partition that stays with its member. Within one session, a rebalance now drops only the positions of the partitions that left, and a new session still clears them all. A partition that moved away and back has a new owner, and the server refuses a poll from its old position. A committing poll now drops the position on that refusal, like a poll that commits nothing, so the partition resolves its strategy again.
Under commit-on-poll the server commits each batch that it reads. A fetch whose reply was lost left the stored record of its partition at the last delivered message, so a shutdown before the retry stored nothing. The stop also aborted a fetch in flight. In both cases a server that committed the batch kept its offset past messages that the application never got, and the next run skipped them. A committing fetch now carries a guard. Unless the fetch settles with a delivered batch or a refusal, the guard forgets the stored record, and the shutdown stores the last delivered message again. The stop wakes the worker, which then starts nothing new and waits up to offset_drain_timeout for its fetches. A batch that arrives in that time comes back with the undelivered batches and is rewound. After that time the worker is aborted. An aborted fetch drops on its own task, so its guard holds the batch channel open and the stop drains the channel only after every guard is gone. The rustdoc no longer claims that only a lost reply to the first Next poll of a partition can skip its batch.
The server turns a deferred Next poll into an explicit offset when it admits the poll, and the poll can then wait up to max_wait. An offset that another session stored during the wait did not reach the waiting poll. After a rewind, the poll read from its old start, and its commit moved the offset past the rewound messages, so no poll read them again. After a store ahead, it returned messages that the store marked as read. A waiting Next poll now compares its start with the start that the current stored offset gives, before each read and before its commit. If the two differ, the server refuses the poll with TransientNotAccepted. Nothing was applied, so the client sends the poll again, and the server resolves Next from the new offset, as it does for a plain poll. Resolving Next again at each read does not cover a store that lands during a disk read, and the waiter counts its selection from a fixed start.
A commit on a watched partition queued every deferred poll that waited on it. Each probe then validated its poll in full: the session, the user, the permissions, the metadata and the partition state. For a group consumer, it also scanned the group members. A poll whose min_count the new messages could not reach then only parked again. With many batching polls on a busy partition, this work ran in front of every inbound frame of the shard. The commit now reads the history, visibility and view of the partition once. If a read could not finish a poll, and the poll keeps its history and view, the poll stays parked without a probe. It moves its bound to the new visibility, as its probe did. Every other poll is queued as before. A probe still validates each poll before its read and before its commit. Maintenance and the deadline still probe the polls that stay parked.
A commit woke each parked readiness wait on its partition, and the probe ran every check on every target of the wait. A wait on 1000 partitions cost 1000 evaluations per commit, and maintenance probed 16 waits per tick whatever their size. A probe now evaluates the targets whose partitions changed, plus a window that maintenance assigns by counting targets, and checks the caller once per wait. A reply can leave out a next partition whose stored offset moved back without a commit. The client waits on that partition again, and admission evaluates every target.
A deferred disk read walked with a third of its 16 MiB reservation and ignored the request's max_bytes. The budget admitted a second 1 MiB chunk, so a poll at the SDK default of 1 MiB fetched nearly 2 MiB, the reply kept 1 MiB, and the next poll read the rest again. The walk now takes the reply's byte limit, min(max_bytes, max_read_bytes) minus the response header. It sizes each read to what the reply can still keep plus the run in front of the first match, and it stops once its selection fills that limit. A first record above max_bytes is still read whole, so the SDK can fetch it alone. Plain polls pass no limit and read as before. The reservation per read is unchanged.
A cancel (command 108) can reach the server while its deferred poll still waits in the connection queue. cancel_deferred_request returns true even when nothing matched, so the queue removal in enqueue_client_request is the only step that keeps the poll from running later. No test covered it. The queue test never sent a cancel, and the integration test cancels a parked poll. Without the removal, the poll would park with nothing left to abort it, and it would commit a batch that no client waits for. The new tests queue an auto-commit poll behind a full cap and send its cancel the way the transport does. The poll leaves the queue and never reaches its owner. A cancel of another request leaves it queued until the cap admits it. The deferred poll frame of the existing queue test moves into a helper that both tests use.
A partition primary that had no free relay slot refused a relayed deferred poll with 429 or 503. The body id was the HTTP status and the body code named the relay cap. The Rust and Java deferred clients back off only on error 58 or 90. Rust reads the body id and Java reads the body code. So a full relay budget failed the poll at once, but the SDK README says that clients back off on a capacity refusal. A refused deferred relay now carries the id and code of DeferredPollCapacityExceeded (90). The status, the reason and Retry-After do not change, and a refused immediate relay keeps its old body. The SDK README names this case.
A refused readiness wait marks the route of every partition in its group stale, and one wait covers up to 1000 partitions. The driver then looked the routes up one after another before it waited again. A primary that held 333 of 1000 partitions on 3 nodes cost 333 round trips, and every consumer of the group stalled that long in a view change. The lookups now run 16 at a time. Each group keeps the order of the partitions, so a consumer does not see a changed group and restart its waits, and the first failed lookup in that order is still the error. No request returns the routes of a whole topic, so each partition still costs one lookup.
poll_next keeps the batch of a commit-on-poll fetch, because the server already committed it. A new fetch generation or assignment must not clear that batch from the buffer, and a batch that carries an older generation must still reach the application. poll_next also skips the messages up to the last delivered offset when a batch reads them again, unless replay is on. No test failed when one of these checks was removed. The new tests call poll_next on a buffered batch across a generation change and an assignment change, on a committing fetch whose generation changes while it is in flight, and on a batch that reads delivered messages again, with and without replay.
IggyConsumer::poll_next fails with InvalidConfiguration until init() runs. Earlier releases polled without init(), so a loop that continues on errors now spins without a single log line. The first poll before init() logs a warning that names the consumer. Later polls skip the warning, and every poll still fails with the same error.
The consumer defaults moved to AutoCommit::Disabled and Next, but the stream-builder path gives no signal. A MessageConsumer gets no handle to store offsets, and the shutdown warning never fires when the consumer is dropped inside a task, so an app on the default config reads from offset 0 on every start without a log line. consume_messages now warns once at its start for Disabled with Next, sharing the shutdown check. IggyConsumerConfig::with_auto_commit lets a default config set a mode, and the stream-consumer and stream-basic examples keep committing on poll.
The DeferredPollRequest doc said that the owner shard sets both deadlines when the request arrives. partition_read runs on the origin shard and stamps them from the shared bus clock before it routes the request to the owner.
Every deferred disk read reserved deferred_poll_max_read_bytes, 16 MiB by default, whatever its max_bytes. So a shard with the default 64 MiB budget ran four deferred disk reads at once, even when each asked for 1 MiB. A read now reserves three times its max_bytes, one part each for its snapshot, disk selection and index loading, but no more than deferred_poll_max_read_bytes. A part is never smaller than the 1 MiB disk chunk, because one read can take a 1 MiB chunk whatever its max_bytes, so small polls cannot start more reads than the budget holds. At the defaults, 21 reads fit one shard. The disk walk still gets a third of the reservation, so a 1 MiB poll now walks with 1 MiB of buffers instead of 5.3 MiB. The gauge moves by the size of each reservation. Reads now differ in size, so the pump wakes and the service loop starts the head of the refused queue only when that read fits. A smaller read behind it keeps its place instead of overtaking it.
Deferred polls replaced the sink polling loop, and the runtime only logged a warning for a stream that still set poll_interval. A sink that batched with it, such as Delta with 3 s, fell back to min_count 1 and max_wait 1 s and made many small commits. One log line at setup was the only signal. Sink setup now refuses such a stream on the path that refuses an invalid max_wait. The sink fails to start with an error that names max_wait and min_count, and the other connectors keep running. The field stays in the config, because serde ignores an unknown field without notice. The in-repo sink configs and the runtime README drop the setting.
All sinks shared one consumer client. Its progress lane per partition and its 64 deferred requests were shared too, so a sink that waited with min_count and max_wait held back the polls and offset stores of other sinks on the same partition, and about 16 sink consumers on a cluster used up the requests of all sinks. A restart also rejoined the group as the same client, so a stopped instance that left late removed the new member. Each sink start now connects its own client. Its consume tasks share it, and the task that ends last logs it out and shuts it down, because a consumer shutdown can still rewind offsets after the 5 s stop wait. A start that fails after the connect logs its client out at once. A closed connection alone keeps its session, and with it the group members of earlier streams, until the session times out. The restart path connects before the plugin starts, so a refused connection leaves no plugin instance behind.
The README said that consumers on one partition share its progress lane, but not what that costs a consumer's own offset calls. They wait while a progress poll is parked there, and a wait close to the 30 s operation timeout makes them fail before they are sent.
hubcio
force-pushed
the
feat/deferred-polling-pr
branch
from
October 27, 2026 00:13
9a8c3ac to
9f87617
Compare
numinnex
approved these changes
Oct 27, 2026
spetz
approved these changes
Oct 27, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Idle consumers repeatedly poll quiet partitions. Park requests until
data arrives or the wait expires, with bounded deadlines, reply
sizes and capacity.
Default Rust consumers to deferred polling with manual commits;
expose bounded polling through Java TCP and HTTP. Route readiness
waits across partition leaders, preserve cancellation order and
refresh budgets on each attempt.
Keep progress ordering separate from producers, bound shared
client admission, and distinguish capacity pressure from routing
failures. Preserve retained offset operations and report uncertain
delivery without replaying progress polls.
Tie auto-commit fetches to demand, preserve partition positions and
document lost-reply uncertainty. Cover limits, oversized records,
cancellation, retries, session changes and cross-leader delivery.
BREAKING CHANGE: Rust consumers default to AutoCommit::Disabled
instead of committing on poll, and the stream builder starts at
PollingStrategy::next() instead of last(). The Python consumer_group
and the PHP consumerGroup take the same auto-commit default. An
application that never stores offsets reads everything again after a
restart, and a new stream-builder group starts at offset zero.
poll_options replaces the consumer builder's poll_interval and
without_poll_interval methods, the IggyConsumerConfig builder's
polling_interval setter and the IggyConsumerConfig::polling_interval
getter. IggyConsumerConfig::new, IggyConsumerConfig::from_stream_topic
and IggyStreamConfig::from_stream_topic take DeferredPollOptions
instead of a polling interval, and IggyConsumer::poll_next fails with
InvalidConfiguration until init() runs. The Python consumer_group
drops poll_interval and makes the arguments after
auto_join_consumer_group keyword-only, and the PHP consumerGroup
refuses a non-null $poll_interval_micros. High-level consumers and
connector sinks need a server that supports deferred polls (command
106), readiness waits (107) and cancels (108). Over HTTP they need
GET /streams/{stream_id}/topics/{topic_id}/messages/deferred. They do
not fall back to plain polls. The connectors runtime ignores the sink
poll_interval setting: sinks wait with max_wait and min_count, which
default to 1 s and 1 message.