merlimat opened a new pull request, #26131:
URL: https://github.com/apache/pulsar/pull/26131
### Motivation
The v5 scalable consumers exposed async receive via
`CompletableFuture.supplyAsync(() -> blockingReceive())`. That fakes async by
parking a `ForkJoinPool.commonPool()` worker for the whole wait:
- the no-timeout `AsyncQueueConsumer.receive()` blocks a pooled thread
**indefinitely** (the API has no timeout variant);
- it is **uncancelable** — `future.cancel(true)` never interrupts the running
`supplyAsync` task, so a caller that gives up still leaks the parked
thread;
- completion happens on the commonPool. A handful of idle async consumers can
starve the JVM-wide commonPool.
The delivery side is already fully async (messages arrive on netty IO
threads); only the handoff to a waiting `receiveAsync()` future was missing.
### Modifications
- Add `V5ReceiveQueue`, modelled on v4 `ConsumerBase`: a buffer of ready
messages plus a queue of pending receive futures, both confined to one
pinned executor per consumer so a message and a waiter can't cross (no
locks,
no lost wakeups). `receiveAsync` parks no thread, is cancelable, and
honours
timeouts via the client timer; futures complete off the netty IO thread.
Blocking `receive()`/`receiveMulti()` are built on the same primitive and
block only the caller's own thread. `close()` now fails pending receives
instead of leaving blocked callers hung forever (the old
`LinkedTransferQueue.take()` never woke on close).
- Rewire all five consumers (`ScalableQueue`/`Stream`/`Checkpoint` + the two
`MultiTopic` wrappers) and their async views onto it, removing every
`supplyAsync`.
### Verifying this change
- New unit test `V5ReceiveQueueTest` (13 cases): buffering, hand-off in both
directions, FIFO order, timeout→null, message-beats-timeout, a cancelled
waiter not swallowing a message, `close()` failing pending receives, and
`receiveMulti` full/partial/empty.
- Full `pulsar-client-v5` suite passes (71 tests); checkstyle passes.
### Note
Touches `closeAsync()` in the two multi-topic consumers, which also appears
in
the retry-timer fix PR; the two are logically independent and may need a
trivial rebase depending on merge order.
--
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]
