merlimat opened a new pull request, #26151:
URL: https://github.com/apache/pulsar/pull/26151
### Motivation
Follow-up to #26131. The per-segment receive loops re-arm unconditionally
into
an unbounded `V5ReceiveQueue`, so if the application stalls, the buffer grows
without bound: the v5 layer keeps draining the v4 consumers, which keep
issuing flow permits to the broker. The v5 layer effectively defeats the v4
consumer's `receiverQueueSize` flow control end-to-end.
### Modifications
Restore backpressure, modelled on v4 `MultiTopicsConsumerImpl`'s pause/resume
of sub-consumers:
- `V5ReceiveQueue.offer()` now returns a `CompletableFuture<Void>` capacity
signal. It completes immediately while the buffer has room; once the buffer
reaches the high watermark (`receiverQueueSize`) the producer is parked
until the consumer drains it to half. v4's fairness clause is carried over:
once any producer is parked, further offers park at the low watermark too,
so active segments can't hold the buffer above the resume threshold and
starve parked peers.
- A new `MessageSink` interface carries the capacity future across the
single-topic/multi-topic sink boundary, so the shared mux's fullness pauses
every per-topic segment loop.
- Each per-segment receive/read loop re-arms only when the capacity future
completes. A paused loop lets the underlying v4 consumer's receiver queue
fill and stop issuing permits — backpressure reaches the broker.
- Fast path: while the buffer is comfortably below the watermark, `offer()`
grants capacity with a shared pre-completed future decided on the caller
thread — the fast-consumer hot path pays no per-message allocation and no
serialized executor hop. The size snapshot may lag by the offers still in
flight, so the effective bound is `receiverQueueSize` + one message per
producer.
- Resume posts each capacity grant as its own executor task (as v4 does)
rather than completing inline, so a wide release of parked segments doesn't
stall queued user receive completions behind O(#segments) re-arm work.
- `close()` releases parked producers so their loops observe the close.
### Verifying this change
7 new `V5ReceiveQueue` unit tests: fast-path immediate grant, immediate
completion below the watermark, parking at the high watermark and resume on
drain, fairness parking of a second producer above the low watermark, resume
through the `receiveMulti` and timed-receive drain paths, and `close()`
releasing parked producers. Full `pulsar-client-v5` suite passes (96 tests).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
