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

[lhotari]

Search before asking

• I searched the 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:

private void appendToDispatchChain(long segmentId, Consumer<Producer<T>> 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):

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.

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

Producer<byte[]> 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<byte[]> a = p.async();
byte[] payload = new byte[1024];
List<CompletableFuture<MessageId>> 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 ([fix][client] Apply no-memory-limit producer queue defaults at producer creation #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

• [feat] PIP-468: V5 dead letter queue with scalable DLQ topic #25652 — PIP-468 V5 DLQ, which introduced appendToDispatchChain (ebec5cea521).
• [Bug] Pulsar-perf producer OutOfDirectMemoryError with slow broker #26340 / [fix][cli] Restore the 64M default client memory limit in pulsar-perf #26341 / [fix][client] Apply no-memory-limit producer queue defaults at producer creation #26342 — pulsar-perf OutOfDirectMemoryError and the producer memory-limit default work; pulsar-perf already sets blockIfQueueFull(true) on the V5 async path.
• [fix][client] Fix deadlock when sending chunked messages with BlockIFQueueFull enabled #17795 — earlier deadlock involving blockIfQueueFull (chunking); [fix][client] Fix producer thread block forever on memory limit controller #21790 — blocked forever in reserveMemory.
• [Bug] Producer send admission control blocks Pulsar's own internal threads, and there is no non-blocking backpressure mode #26343 — producer send admission control blocks the calling thread: the v4-level root cause, and the AsyncSemaphore fix vehicle.