[Bug] V5 producer can self-deadlock a Netty IO thread: the segment dispatch chain runs sends on the connection's event loop

### Search before asking

- [x] I searched the [issues](https://github.com/apache/pulsar/issues) and found nothing similar.

### Problem

With the V5 client, `blockIfQueueFull(true)`, and async sends, all sends queued while a segment producer is being created execute **on that connection's Netty event-loop thread**. If they collectively exceed the client memory limit, one of them parks in `MemoryLimitController.reserveMemory`. That memory can only be released by a send receipt delivered **by the same event loop**, and the parked thread holds the `ProducerImpl` monitor that `ackReceived` needs. This is a self-deadlock, not merely a stalled event loop.

There is no escape:

- Neither blocking call takes a timeout.
- The send-timeout task (`ProducerImpl.run(Timeout)`, `:2316`) immediately does `synchronized (this)` (`:2323`), so the client's **shared `HashedWheelTimer` worker** blocks on the same monitor — taking down send timeouts and consumer ack timeouts for *every* producer and consumer on that client.
- Even if the timer got the monitor, `failPendingMessages(cnx(), te)` with a live `cnx` defers the release back onto the blocked event loop.
- `numIoThreads` defaults to `availableProcessors()` and connections are multiplexed onto that fixed group, so one wedged thread also stalls unrelated producers and consumers.

There is a **second, independent defect** in the same code, reachable regardless of `blockIfQueueFull`: while a segment producer is being created, V5 applies *no* admission control at all.

### Concrete evidence

**The chain is built with a bare `thenApply`** — `pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableTopicProducer.java:370-387`:

```java
private void appendToDispatchChain(long segmentId, Consumer> dispatchOp, ...) {
synchronized (dispatchLock) { // :373
var prev = dispatchChains.computeIfAbsent(segmentId,
id -> getOrCreateSegmentProducerAsync(id)); // :374-375
var next = prev.thenApply(producer -> { // :376 NO executor
dispatchOp.accept(producer);
return producer;
});
...
dispatchChains.put(segmentId, next); // :385
```

**The chain head is completed on a `ClientCnx` event-loop thread.** `getOrCreateSegmentProducerAsync` (`:562`) resolves to `PulsarClientImpl`'s producer-creation future, returned with no intermediate stage; its single completion point is `ProducerImpl.java:2234`, inside `resendMessages` (`:2218`):

```java
private void resendMessages(ClientCnx cnx, long expectedEpoch) {
cnx.ctx().channel().eventLoop().execute(() -> { // :2219
synchronized (ProducerImpl.this) { // :2220
...
producerCreatedFuture.complete(ProducerImpl.this); // :2234
```

`ClientCnx` is a `ChannelInboundHandlerAdapter`, so this is a `pulsar-client-io-*` thread — and the monitor is held for the duration.

**All queued links then fire in one burst on that thread.** Standard `CompletableFuture` semantics: a non-async dependent runs on the completing thread, and `postComplete` unwinds iteratively, so there is no natural bound. Confirmed executably with the exact chain shape from `:376` — 200,000 chained links all ran on the single completing thread. During that burst not one byte reaches the socket, because `ProducerImpl.processOpSendMsg` defers the write with `eventLoop().execute(...)`, which — called *from* the event loop — only enqueues a task.

**Each link reaches the blocking call with no thread handoff.** `ScalableTopicProducer.java:326-328` (`.sendAsync()` at `:328`) → `ProducerImpl.internalSendAsync` (`:404`) → `sendAsync` → `canEnqueueRequest` (`:576`) → `reserveMemory` (`:1130`) → `condition.await()` (`MemoryLimitController.java:116`).

V5 never has a pending-message semaphore — `maxPendingMessages` has no V5 setter and stays at `DEFAULT_MAX_PENDING_MESSAGES = 0`, and `ProducerImpl.java:216-220` only builds the semaphore when it is `> 0`. **The client memory limit (default 64 MiB) is V5's only admission control.**

**The release path is the same thread.** `releaseMemory` is reached only from `releaseSemaphoreForSendOp` (`:1439-1444`), called from `ackReceived` (`:1389`); `ackReceived` (`:1350`) is invoked from `ClientCnx.handleSendReceipt` — that channel's event loop — and needs `synchronized (this)` (`:1352`), the monitor held since `:2220`.

**The class's own javadoc states the assumption being violated** (`ScalableTopicProducer.java:79-80`): "*Each async send appends a link whose sole job is to call `v4Producer.sendAsync(...)` (**fast, synchronous queue insert**)*", and `:70-73`: "*callers running on a netty IO thread can chain on the future asynchronously instead of forcing a blocking `.get()` (which would deadlock against the segment producer's own lookup response, processed on the same IO thread)*". The `.get()` hazard was designed around; the `sendAsync` one was not.

**Retry re-arms it.** `dispatchSendAttempt`'s retry (`:314-321`) removes both the `segmentProducers` and `dispatchChains` entries, so the chain head reverts to a pending creation future — now at full steady-state send rate, after up to `Math.min(100 * (attempt+1), 500)` ms of backoff (`:317`) over up to `SEND_RETRY_MAX_ATTEMPTS = 10` attempts (`:55`).

**Steady state has a second hazard:** once the head is complete, each `prev.thenApply(...)` runs inline on the *calling* thread inside `synchronized (dispatchLock)` (`:373`) — and `dispatchLock` covers **all** segments, so one blocked link freezes dispatch for every segment of that producer.

### Second defect: no admission control during segment-producer creation

`sendInternalAsync` (`:277-289`) does three things — allocate `userFuture`, add it to `inFlightSends`, append a chain link. Nothing reserves memory, acquires a permit, or bounds the queue; memory is only consulted once the v4 `ProducerImpl` exists.

Nothing bounds the accumulation except heap. Each queued link retains the captured `key`, `value` (full payload), `properties`, `eventTime`, `sequenceId`, `deliverAfter`, `deliverAt`, `replicationClusters`, `txn` and the `userFuture`; `inFlightSends` is an unbounded `ConcurrentHashMap.newKeySet()`. **This holds regardless of `blockIfQueueFull`** — with `false`, the burst instead fails everything above the limit with `MemoryBufferIsFullError` in one shot, which is its own usability bug.

The window is not small: a cold segment producer is a partition-metadata lookup + connect + `CommandProducer` round trip, plus broker-side topic auto-creation (ledger creation, routinely 100 ms+); the retry path adds up to 4 s of cumulative backoff at full send rate.

*(Adjacent, out of scope: `dispatchChains.put(segmentId, next)` at `:385` stores the `thenApply` result, not the `exceptionally` result. If creation fails with something that is not an `isSegmentGoneError`, both maps retain the faulted future and every subsequent send to that segment fails forever with the stale error.)*

### Reproduction

Pre-existing on master; introduced with the dispatch chain in `ebec5cea521` (PR #25652, PIP-468) and untouched since.

```java
PulsarClient v5 = PulsarClient.builder() // org.apache.pulsar.client.api.v5.PulsarClient
.serviceUrl(brokerUrl)
.memoryLimit(MemorySize.ofBytes(1 << 20)) // 1 MiB
.build();

Producer p = v5.newProducer(Schema.bytes())
.topic("persistent://public/default/repro") // a regular topic is enough
.blockIfQueueFull(true)
.batchingPolicy(BatchingPolicy.ofDisabled())
.create(); // segment producer NOT yet created

AsyncProducer a = p.async();
byte[] payload = new byte[1024];
List> fs = new ArrayList<>();
for (int i = 0; i < 4000; i++) { // 4 MiB >> 1 MiB, queued in microseconds
fs.add(a.newMessage().value(payload).send()); // first call triggers lazy creation
}
CompletableFuture.allOf(fs.toArray(new CompletableFuture[0])).get(60, SECONDS); // never completes
```

Preconditions: (1) the V5 client — `ScalableTopicProducer` also backs plain `persistent://` topics via synthetic legacy layouts, so this is not scalable-topic-only; (2) `blockIfQueueFull(true)`; (3) async sends — the sync path (`:204`) does a blocking `.get()` on the app thread and never touches the chain; (4) more than the memory limit queued before the segment producer is ready. Keep the default (non-exclusive) access mode so segment producers are created lazily — `requiresExclusiveAttach()` triggers eager attach at `create()`, closing the window.

`pulsar-perf produce` satisfies 1–3 out of the box (`pulsar-testclient/.../PerformanceProducer.java:477-482` builds a V5 `ProducerBuilder` with `.blockIfQueueFull(true)` and sends via `p.async()`); its *default* rate is too slow for 4, but any realistic benchmark rate is not.

Expected thread dump:

- `pulsar-client-io-*`: `Unsafe.park` ← `MemoryLimitController.reserveMemory` (`:116`) ← `canEnqueueRequest` (`:1130`) ← `ProducerImpl.sendAsync` (`:576`) ← `CompletableFuture$UniApply.tryFire` ← `ProducerImpl.lambda$resendMessages$…` — **locked `ProducerImpl@…`**
- `pulsar-timer-*`: `BLOCKED` on that same `ProducerImpl@…` in `ProducerImpl.run(Timeout)` (`:2323`)
- No `CommandSend` on the wire, confirming no receipt can ever arrive.

### Proposed solution

**(A) Complete the chain head off the event loop — one hop, not one per message.** Seed the chain as `getOrCreateSegmentProducerAsync(id).thenApplyAsync(Function.identity(), producerDispatchExecutor)` instead of the raw future at `:375`.
*Ordering:* preserved. The chain already serializes by construction — link N+1's function is only scheduled once link N's has returned, so only one link is ever runnable; any executor preserves order. A dedicated single thread is for confinement, not ordering.
*Cost:* one thread hop per chain head; steady-state appends still run inline, so no per-message latency tax.
*Trade-off:* the executor must be **dedicated per producer** (or per segment). Routing it to a shared pool — e.g. `client.getInternalExecutorService()`, already used at `:481`/`:533` — just relocates the wedge onto a shared thread.
*Insufficient alone:* converts the deadlock into unbounded queueing, since the app thread now never blocks. Must be paired with (B).

**(B) Move admission control up into `sendInternalAsync`, before the chain append.** Semantically the correct fix: `blockIfQueueFull(true)` is documented as blocking the caller's send call, which today it does not do at all during segment-producer creation. Reserve V5-side and create the v4 segment producers with accounting disabled (or add a "pre-reserved" flag on the v4 send path) so bytes are not double-counted.
*Trade-off:* touches the v4/v5 boundary and needs a matching release on every terminal path (ack, fail, timeout, close, retry re-dispatch — note the retry at `:314-321` re-dispatches the *same* message, so the reservation must not double-count).
*Also fixes the `blockIfQueueFull(false)` case:* the caller gets `MemoryBufferIsFullError` promptly instead of silently buffering to OOM.

**(C) Async backpressure — never block anywhere.** Best long-term fit for an async-first V5 API: non-blocking capacity acquisition (`AsyncSemaphore` / `AsyncDualMemoryLimiter` from `pulsar-common` already exist and `pulsar-client` already depends on them), with `blockIfQueueFull(true)` meaning "the send future is delayed until capacity exists" rather than "a thread parks". Largest change, and it makes `blockIfQueueFull` mean something different in V5 than v4 — but V5 is a new API surface, so this is the moment to define it. See #26343.

**(D) Cheap safety nets, orthogonal to the above.**
- **Fix the V5 javadoc.** `pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/ProducerBuilder.java:95-96` says `blockIfQueueFull` "Default is `true`". The actual default is `false` (`ProducerConfigurationData.java:87`) and `ProducerBuilderV5` never overrides it (its only writer is the setter at `:131-134`). The v4 javadoc is correct. **The doc must be corrected to `false`, not the code changed to `true`** — making it `true` would make this deadlock the default. The javadoc should also state that V5's only queue bound is the client memory limit.
- **Fail fast instead of hanging:** in `canEnqueueRequest`, when `conf.isBlockIfQueueFull()` and the current thread is a Netty event-loop thread, complete exceptionally rather than park. There is currently **no `inEventLoop` guard anywhere** in `pulsar-client` or `pulsar-client-v5`.
- **Future work note:** extending the pending-message-limit defaults (#26342) to `createSegmentProducerAsync` would give V5 producers a semaphore, adding `semaphore.get().acquire()` (`:1128`) as a *second* blocking call on the same wedged thread. Land (A)/(B)/(C) first.

### Scope & compatibility

- (A) and (D)'s doc fix are internal/documentation only — bug-fix scope, no PIP.
- (B) changes *when* a V5 send is admitted and can make `blockIfQueueFull(false)` fail sends that are today buffered silently. It restores documented behaviour rather than changing it; still worth release notes.
- (C) redefines `blockIfQueueFull` semantics for V5 → **PIP required**.
- No wire-protocol or broker-side impact. V5 is a new API surface, so there is no released-behaviour compatibility constraint beyond what has already shipped.

### Related

- #25652 — PIP-468 V5 DLQ, which introduced `appendToDispatchChain` (`ebec5cea521`).
- #26340 / #26341 / #26342 — pulsar-perf `OutOfDirectMemoryError` and the producer memory-limit default work; `pulsar-perf` already sets `blockIfQueueFull(true)` on the V5 async path.
- #17795 — earlier deadlock involving `blockIfQueueFull` (chunking); #21790 — blocked forever in `reserveMemory`.
- #26343 — producer send admission control blocks the calling thread: the v4-level root cause, and the `AsyncSemaphore` fix vehicle.