apache / apache/pulsar

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

Open
#26,343 0 comments 0 reactions 1 assignee Claimed by @lhotari View on GitHub
Dominant language
Java
Stars
15.3k
Forks
3.8k
Avg merge
1d 14h
Merged PRs (30d)
160

Description

### Search before asking

- [x] I searched the [issues](https://github.com/apache/pulsar/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`:

```java
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` 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 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: #16837, #17276, #17936, #21790, #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

- #26340 / #26341 / #26342 — pulsar-perf `OutOfDirectMemoryError` and the producer memory-limit default work (user-visible symptom of this area).
- #20663 — set the perf-tool memory limit default to 0.
- #21790 — blocked forever inside `reserveMemory`; #17795 — chunking deadlock with `blockIfQueueFull`.
- #24229 — REST-API publisher bypasses producer rate limiting / memory limits.
- #25699 — document Pulsar's concurrency and memory-backpressure model.
- #23643 (closed) — the doc clarification that produced the current `sendAsync` javadoc.
- PIP-74 — original client memory-limit design. PIP-442 — `AsyncSemaphore` / `AsyncDualMemoryLimiter`.
- Companion issues: #26344 (V5 Netty IO thread self-deadlock), #26345 (shared `MemoryLimitController` defects), #26346 (configurable memory limit for broker/proxy clients).

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.