[Bug] Producer send admission control blocks Pulsar's own internal threads, and there is no non-blocking backpressure mode

[lhotari]

Search before asking

• I searched the issues and found nothing similar.

Motivation

A producer has exactly two admission-control behaviours, selected by blockIfQueueFull:

• false (default): a send exceeding maxPendingMessages or the client memory limit fails with ProducerQueueIsFullError / MemoryBufferIsFullError.
• true: the send parks the calling thread until capacity is available.

There is no third option — "delay the send until capacity exists, without occupying a thread" — which is what an async client needs.

To be clear about what this is not: caller-blocking under blockIfQueueFull(true) is documented behaviour (ProducerBuilder#maxPendingMessages/#blockIfQueueFull javadoc, and Producer#sendAsync points at it), and PIP-74 designed it that way. For an application thread it is a legitimate choice. The defects are:

(a) The blocking is not confined to application threads. sendAsync is routinely invoked from continuations of Pulsar-owned futures, and those continuations run on Pulsar-owned threads — the transaction pinned executor, the listener executor, and Netty IO threads. Blocking there violates CODING.md rule 2 ("Never block on async/event-loop threads") and is documented nowhere.

(b) Applications wanting real backpressure must reimplement it. With the default false, the only supported strategy is catch-and-retry-with-backoff on two exception types. The public API exposes no permit, readiness signal, or capacity future; ProducerStats.getPendingQueueSize() is a gauge and MemoryLimitController is impl-only.

Concrete evidence

Single admission point — pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java:1124-1153:

private boolean canEnqueueRequest(SendCallback callback, long sequenceId, int payloadSize) {
    try {
        if (conf.isBlockIfQueueFull()) {
            if (semaphore.isPresent()) {
                semaphore.get().acquire();                                    // :1128  blocking
            }
            client.getMemoryLimitController().reserveMemory(payloadSize);     // :1130  blocking
        } else {
            ... ProducerQueueIsFullError (:1134) / MemoryBufferIsFullError (:1141) ...

MemoryLimitController.reserveMemory (MemoryLimitController.java:107-122) takes a lock and loops on condition.await() at :116. semaphore is a fair Semaphore, created only when maxPendingMessages > 0 (ProducerImpl.java:216-220). Neither call takes a timeout. canEnqueueRequest is reached synchronously from sendAsync (ProducerImpl.java:576, via internalSendAsync :404-421) with no thread handoff on the path.

sendTimeout cannot bound this wait: op.createdAt is stamped in OpSendMsg.create (:1661, :1675, :1695), i.e. after the permit is taken, and run(Timeout) (:2316) only scans pendingMessages and the batch container. A parked send is invisible to it.

Which threads actually execute canEnqueueRequest:

Thread	Path	Status
App thread via send()	ProducerBase.send → internalSendAsync(...).get()	Documented, expected
App thread via sendAsync()	direct	Documented
Transaction pinned executor	internalSendWithTxnAsync (:511-522) does registerProducedTopic(topic).thenCompose(ignored -> internalSendAsync(message)) at :519-520 with no executor; that future completes inside TransactionMetaStoreHandler's internalPinnedExecutor	Undocumented, violates rule 2
Listener (external pinned) executor	any MessageListener calling sendAsync — the standard consume-transform-produce shape	Undocumented
Netty IO thread	any sendAsync chained off a Pulsar future without an explicit executor. Send futures complete from ClientCnx.handleSendReceipt → ackReceived (:1350); producer-creation futures complete at :2234 inside eventLoop().execute(...) (:2219)	Deadlock-capable — see companion V5 issue

The blockIfQueueFull javadoc's own example is sendAsync(...).thenAccept(...) in a loop — precisely the shape that migrates work onto the IO thread.

In-tree call sites selecting the blocking branch: pulsar-functions/instance/.../ProducerBuilderFactory.java:75 (with sendTimeout(0), deliberately — the comment cites deadlock avoidance), pulsar-functions/worker/.../WorkerUtils.java:476, pulsar-broker/.../loadbalance/extensions/channel/ServiceUnitStateTableViewImpl.java:78-79, pulsar-testclient/.../PerformanceProducer.java:482.

Note ServiceUnitStateTableViewImpl specifically: it sets maxPendingMessages(500) + blockIfQueueFull(true) and its put(String, ServiceUnitStateData) returns a CompletableFuture<Void> while sendAsync() can park the calling thread once 500 publishes are outstanding — a rule-2 violation in broker code.

Conversely ConsumerImpl.java:2458/:2527 explicitly set blockIfQueueFull(false) on the DLQ and retry-letter producers because those sends originate on internal threads. The codebase already treats this as a known hazard, handled by hand, per producer.

Proposed solution

Add a third mode: admission control returning a future instead of parking a thread, so the send future simply completes later. Both existing blockIfQueueFull modes keep their semantics.

The primitive already exists and is already on the compile classpath (pulsar-client/build.gradle.kts:48 → api(project(":pulsar-common"))):

• org.apache.pulsar.common.semaphore.AsyncSemaphore / AsyncSemaphoreImpl — CompletableFuture<AsyncSemaphorePermit> acquire(long, BooleanSupplier isCancelled), strictly FIFO, bounded waiter queue, per-request timeout, cancellation, and update() for resizing a held permit (which maps onto the batch container's reserve-then-resize pattern).
• AsyncDualMemoryLimiter / AsyncDualMemoryLimiterImpl — heap/direct split with withAcquiredPermits(...).

Both landed with PIP-442 and have zero client-side users today (only TopicListMemoryLimiter, TopicListService, LookupProxyHandler). The missing piece is producer-side integration, not the primitive.

Two gaps to close first: AsyncSemaphoreImpl.validatePermits throws IllegalArgumentException synchronously when permits > maxPermits, whereas MemoryLimitController.tryReserveMemory deliberately lets one oversized request through (MemoryLimitController.java:71-75); and it has no equivalent of registerTrigger (:161-167), used for receiver-queue shrinking and the shared-resource path.

Alternatives considered

• Redefine blockIfQueueFull=true as "async delay". Smallest API surface, but silently changes documented semantics and breaks callers relying on caller-thread rate limiting (Functions).
• Fail fast when the calling thread is an event-loop thread. There is currently no inEventLoop guard anywhere in pulsar-client or pulsar-client-v5. Converts a silent permanent hang into a diagnosable error — cheap, and a useful mitigation, but not a fix: it does not help the transaction/listener executor cases, where blocking is bad rather than fatal.
• Document the hazard only. A strict improvement over today, but leaves applications with no way to apply backpressure without blocking.

This is genuinely hard — what a fix must preserve

1. Ordering and sequence-ID assignment. msgIdGenerator++ happens at serialization time inside synchronized (this) (:696-706, :745-754), i.e. after the permit. If waits complete out of call order, sequence IDs follow the reordering and so does broker dedup. FIFO inside the semaphore is necessary but not sufficient — a per-producer dispatch chain is needed. (V5 already had to build one; see ScalableTopicProducer.java:78-89.) The two existing limiters already disagree on fairness: the pending-message semaphore is fair, reserveMemory is signal-and-retry and is not FIFO.
2. sendTimeout does not currently cover the wait. Bringing it under sendTimeout is itself a behaviour change; Functions deliberately runs sendTimeout(0).
3. flush() would silently under-cover. flushAsync returns a handle on lastSendFuture (:2445-2457), assigned only on enqueue (:817, :867, :1071). Parked messages would be excluded, so flush() could return before them — impossible today.
4. Waiters need a bound, and they hold unaccounted memory. Today the blocked thread is the bound. Worse, the payload ByteBuf is retained and interceptors have already run before the permit check (:407-410), so every parked message holds a retained buffer that is by construction not counted in the limit it is waiting on — a second, invisible pool.
5. Accounting on failure paths is already fragile. Release points are spread across completeCallbackAndReleaseSemaphore (:1446-1450), releaseSemaphoreForSendOp (:1439-1444), the chunk-failure branches (:658-665, :689-695) and BatchMessageContainerImpl. Prior fixes: [fix][client]Fix client memory limit currentUsage leak and semaphore release duplicated in ProducerImpl #16837, [fix][broker]Fix memoryLimitController currentUsage and MaxQueueSize semaphore leak when batchMessageContainer add message exception #17276, [fix][client] Reserve allocated buffer in BatchMessageContainer on client memory limitation. #17936, [fix][client] Fix producer thread block forever on memory limit controller #21790, [fix][client] Fix failed to close consumer because of the error: param memorySize is a negative value #25805. A fix must survive cancel-, close- and disconnect-while-waiting, plus the chunked path where memory is reserved once and chunks 2..N pass 0.
6. Release happens on the IO thread (releaseSemaphoreForSendOp inside ackReceived, :1389), so a waiter future completes there and the parked send is enqueued on the IO thread. Acceptable only if that continuation stays a fast queue insert.
7. The controller can be shared across clients (SharedResource.MemoryLimitController). A shared waiter queue introduces cross-client starvation, and closing one client must drain only its own waiters.

Scope & compatibility

• No change to existing defaults or to either documented blockIfQueueFull mode is proposed.
• Adding a new mode/knob to ProducerBuilder or ClientBuilder is a public client API change → PIP required.
• The "fail fast on an event-loop thread" mitigation and javadoc additions are bug-fix scope, no PIP; the fail-fast change alters behaviour in a case that currently hangs, so it warrants release notes.
• No wire-protocol, metadata-format or broker-compatibility impact.

Related

• [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 (user-visible symptom of this area).
• [feat][cli] Add command line option for configuring the memory limit #20663 — set the perf-tool memory limit default to 0.
• [fix][client] Fix producer thread block forever on memory limit controller #21790 — blocked forever inside reserveMemory; [fix][client] Fix deadlock when sending chunked messages with BlockIFQueueFull enabled #17795 — chunking deadlock with blockIfQueueFull.
• [Bug] REST API publisher by-passes producer rate limiting / memory limits and lacks back pressure #24229 — REST-API publisher bypasses producer rate limiting / memory limits.
• [doc] Document Pulsar's concurrency model and memory backpressure model #25699 — document Pulsar's concurrency and memory-backpressure model.
• [Doc] Description for Producer SendAsync Needs Updating #23643 (closed) — the doc clarification that produced the current sendAsync javadoc.
• PIP-74 — original client memory-limit design. PIP-442 — AsyncSemaphore / AsyncDualMemoryLimiter.
• Companion issues: [Bug] V5 producer can self-deadlock a Netty IO thread: the segment dispatch chain runs sends on the connection's event loop #26344 (V5 Netty IO thread self-deadlock), [Bug] Shared MemoryLimitController silently disables the client memory limit and drops the client memory metrics #26345 (shared MemoryLimitController defects), [Enhancement] Configurable memory limit for broker, proxy and WebSocket proxy Pulsar clients, shared across all of them #26346 (configurable memory limit for broker/proxy clients).