[fix][client] Apply backpressure in the V5 producer before a send is queued

[merlimat]

Fixes #26470. Also removes the IO-thread self-deadlock described in #26344.

Motivation

A V5 async producer never pushed back on its caller. AsyncMessageBuilder.send() appended a link to the target segment's dispatch chain and returned; the client memory limit, the only bound a V5 producer has, was enforced by the v4 ProducerImpl the message reached later, on whichever thread completed the previous link. That thread is the connection's IO thread, which completes the chain head when the segment producer is created.

Once the chain had fallen behind by a single link, every subsequent send() was a pure append, and under sustained load the chain never catches up: in a standalone reproduction of the chain shape, an application thread appended 10.6M links in 3 seconds while the completing thread executed 2.6M, not one link ran on the application thread, and the backlog peaked at 10.5M. While the IO thread unwinds the chain it cannot run the socket writes queued by processOpSendMsg, so nothing reaches the wire and no acknowledgement ever arrives, which is why the heap dump in the issue shows the limiter at 44% of its budget and not one message published.

Independently, the limit was charged the payload alone, while each pending message retains about 2 KiB of bookkeeping (metadata, send op, callbacks, futures), so for small messages the limit was an order of magnitude optimistic even when a message did reach the v4 producer.

Modifications

• Admission on the caller's thread, before a send is queued (ScalableTopicProducer). Each send is charged its encoded payload plus a fixed 1 KiB per-message allowance against the client memory limit, so the limit bounds the number of pending messages as well as their bytes. With blockIfQueueFull(true) the caller blocks until acknowledgements make room; with false the send fails right away with MemoryBufferIsFullException. The admission happens before the dispatch lock is taken, so a waiting sender never stalls the other segments.
• A send from one of the client's IO threads never blocks. Send futures complete on the IO thread that received the broker's response, as in v4, so a send issued from a continuation runs there. Parking that thread would hold up the acknowledgements that free memory, so at the limit such a send fails fast with MemoryBufferIsFullException whatever the setting. The check iterates the client's event loop group and only runs once a reservation has failed; the fast path is a single reservation.
• The value is encoded on the caller's thread so the reservation is exact and the schema's CPU work stays off the dispatch thread. New internal hook TypedMessageBuilderImpl#encodedValue(EncodeData). Not used for AUTO_PRODUCE_BYTES (which needs the connected producer's schema) or key/value schemas; their payload is charged by byte length.
• Segment producers only account, never block or reject. New internal ProducerConfigurationData#memoryLimitAdmittedUpstream flag makes ProducerImpl#canEnqueueRequest force-reserve. Every v4 release path is unchanged, so the v4 accounting stays balanced on its own; the V5 layer releases its payload share once the message is handed to the v4 producer and the per-message allowance when the send completes, including on the creation-failure, retry and synchronous-send paths.
• The chain head completes off the IO thread that creates the segment producer (option A of [Bug] V5 producer can self-deadlock a Netty IO thread: the segment dispatch chain runs sends on the connection's event loop #26344), so the burst of links queued during creation does not run there. Links no longer block, so a shared executor is fine.
• blockIfQueueFull defaults to true for V5 producers, matching its javadoc. The V5 dead-letter producer opts out, since forwarding runs on the client's own threads, mirroring the v4 dead-letter and retry producers.
• A failure thrown while building or enqueuing one message now fails that send only, instead of leaving the segment's chain permanently failed.
• Javadocs on ProducerBuilder#blockIfQueueFull, PulsarClientBuilder#memoryLimit, AsyncProducer and AsyncMessageBuilder#send describe the actual semantics.

Not changed here: a segment-producer creation failure that is not a segment-gone error still leaves the failed future in place for that segment (the adjacent note in #26344), and the non-blocking capacity-signal mode from #26343 remains a PIP-level change.

Verifying this change

This change added tests and can be verified as follows:

• V5ProducerBackpressureTest (real broker):
  • asyncSendsBlockTheCallerAtTheMemoryLimit: 20,000 async sends of 1 KiB from a tight loop with a 1 MiB limit, never awaited. The number of returned-but-incomplete sends stays under limit / payload + 1 (it reaches about 20,000 without this change), every future completes, and the limiter returns to 0.
  • asyncSendsFailFastAtTheMemoryLimitWhenNotBlocking: with blockIfQueueFull(false) the loop overruns the limit, the rejected sends fail with MemoryBufferIsFullException, the rest succeed, and the limiter returns to 0.
  • sendFromAnIoThreadFailsFastInsteadOfBlocking: fills the limit from inside a send continuation (which runs on an IO thread) and sends again from there; the send fails fast rather than parking the IO thread.
  • syncSendsAreChargedAndReleased and memoryIsReleasedWhenTheSendFails cover the synchronous path and the release on an enqueue-time failure.
• ProducerMemoryLimitTest.testMemoryLimitAdmittedUpstreamOnlyAccounts: a v4 producer with the flag neither rejects nor blocks a send over the limit and still releases the bytes.
• All org.apache.pulsar.client.api.v5.* and org.apache.pulsar.client.impl.v5.* broker tests pass (59 classes, 205 tests), as do the v4 TypedMessageBuilderImplTest, ProducerImplTest, MemoryLimitControllerTest and ProducerMemoryLimitTest.

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

• Dependencies (add or upgrade a dependency)
• The public API: V5 only. blockIfQueueFull now defaults to true, and a V5 send may block the caller at the client memory limit, or fail fast when issued from an IO thread; the javadocs of the V5 builder and async producer are updated accordingly. No v4 API change; the two v4 additions (memoryLimitAdmittedUpstream, encodedValue) are internal hooks.
• The schema
• The default values of configurations: V5 ProducerBuilder#blockIfQueueFull defaults to true (it was false, contrary to its javadoc).
• The threading model: V5 sends are admitted on the caller's thread; the dispatch chain head runs on the client's internal executor instead of the IO thread; send futures keep completing on the IO thread.
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment

[lhotari]

Thanks for picking this up — moving admission onto the caller's thread ahead of the dispatch chain is the right shape, and making the per-segment producers account-only is what keeps the chain itself from ever waiting for memory.

Two things look worth sorting out before this lands, both about when the reservation is given back rather than when it is taken. The cleanup is a dependent of the user-visible future, so it runs after any continuation the caller attached, and it fires on caller-driven completion. That lets a send from a continuation park the client's internal dispatch thread behind its own release, and lets cancel/orTimeout hand the budget back while the message is still queued. Both would go away if this layer released on its own terminal events and completed the caller's future last.

Also flagged a couple of smaller things: some test coverage worth adding while the accounting is fresh, and a latent invariant around the new memoryLimitAdmittedUpstream flag.

[lhotari]

[BUG] Release the reservation before the caller’s continuations run, and fail fast on the dispatch thread too

The reservation is given back by a dependent of the user-visible future, so it runs after whatever the application chained onto that future. CompletableFuture pops dependents LIFO, and this handler is registered at :509 before the future is returned — so a caller's continuation sits above it on the stack and runs first, while this message's payload and its 1 KiB overhead are still charged.

That is harmless until the completing thread is one that cannot afford to wait. admit fails fast only on a Netty IO thread:

pulsar/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableTopicProducer.java

Lines 320 to 338 in 881c175

	private void admit(PendingSend<T> send) throws PulsarClientException {
	long bytes = (long) send.payloadSize + PER_MESSAGE_OVERHEAD_BYTES;
	if (memoryLimit.tryReserveMemory(bytes)) {
	return;
	}
	if (!producerConf.isBlockIfQueueFull()) {
	throw new PulsarClientException.MemoryBufferIsFullException("Client memory buffer is full");
	}
	if (isEventLoopThread()) {
	throw new PulsarClientException.MemoryBufferIsFullException(
	"Client memory buffer is full, and a send from a Pulsar IO thread cannot wait for it");
	}
	try {
	memoryLimit.reserveMemory(bytes);
	} catch (InterruptedException e) {
	Thread.currentThread().interrupt();
	throw new PulsarClientException("Interrupted while waiting for client memory", e);
	}
	}

For the first send to a segment the chain head hops to dispatchExecutor (:176, :604-606), which is a single-threaded slot of the client's internal executor — not an event loop. A v4 send that fails before enqueue completes its callback synchronously (an oversized message takes ProducerImpl.java:599-611), so :552-556 keeps the payload share and the user future is completed inline on that dispatch thread. A continuation that issues a blocking send then parks in reserveMemory, and the release that would free it is the next dependent on the very thread it just parked. With nothing else in flight, it never runs.

memoryIsReleasedWhenTheSendFails already builds this setup — same oversized message, same cold segment. Adding a continuation that sends should turn it into a hang at the timeout.

Releasing before completing the future is necessary but not sufficient on its own: the continuation still runs on the dispatch thread, and any other over-the-limit message queued behind it there keeps usage above the limit. admit also needs to fail fast on the dispatch executor and on the retry delayedExecutor, or user futures need to complete off the client's own threads.

A milder version of the same ordering shows up on the success path: the ack completes the future before the 1 KiB is returned, so a send-on-ack continuation at a tight limit can get a spurious MemoryBufferIsFullException for budget the next dependent is about to release.

[lhotari]

[BUG] Tie the reservation to the send’s own lifecycle rather than to the caller’s future

AsyncMessageBuilderV5.send() hands the caller this future itself, so cancel(), orTimeout() or complete() on it fires the handler at :509 and returns the whole reservation while the message is still queued. Nothing on the dispatch or retry path consults userFuture.isDone().

Two consequences. A per-send orTimeout is ordinary Java — and inviting on a call documented as possibly blocking — and it returns the budget for every timed-out send while the payload and the chain entry stay queued. A stalled broker or a slow segment creation therefore switches the admission bound off for precisely the messages that are backing up.

The accounting can also leak for good. Once an early completion has released the share, a segment-gone failure takes it again here:

pulsar/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableTopicProducer.java

Lines 531 to 540 in 881c175

	Runnable retry = () -> {
	segmentProducers.remove(routedSegmentId);
	dispatchChains.remove(routedSegmentId);
	// The message stays with this layer while it waits for the new layout.
	reholdPayloadShare(send);
	CompletableFuture.delayedExecutor(
	Math.min(100L * (attempt + 1), SEND_RETRY_MAX_BACKOFF_MS),
	TimeUnit.MILLISECONDS)
	.execute(() -> dispatchSendAttempt(send, attempt + 1));
	};

If a later attempt then ends without reaching the v4 producer — retry budget exhausted, or routeMessage throwing at :520-523 — the terminal completeExceptionally is a no-op on the already-completed future, the handler at :509 never runs again, and those bytes stay charged for the life of the client.

Worth noting the exposure is one-directional: the share is CAS-guarded at :350-354, so there is no double release. v4 avoids this shape by releasing from the send-callback path regardless of what the caller does with the returned future. Releasing on this layer's own terminal events and completing the caller's future last would do the same here, and skipping dispatch/retry once it is done would stop the queued work outliving it.

[lhotari]

[TEST] Cover the sending continuation, the retry re-hold, and tighten the outstanding-sends bound

Three gaps worth closing while this is fresh:

• Nothing sends from a continuation completed on the cold-segment path. sendFromAnIoThreadFailsFastInsteadOfBlocking covers the ack/IO-thread case; this test is one .exceptionally(...) away from covering the other one.
• reholdPayloadShare and the retry path (:360-364, :531-540) are the only new bookkeeping with a CAS state machine, and no test seals a segment mid-flight and then asserts currentUsage() == 0.
• asyncSendsBlockTheCallerAtTheMemoryLimit bounds outstanding sends at limit / payload + 1 = 1025, but each send is charged payload + 1 KiB, so the code actually enforces about 513. A regression that dropped PER_MESSAGE_OVERHEAD_BYTES from the charge would still pass — tightening it to the bound really enforced would pin the thing this PR adds.

[lhotari]

[NIT] Make the "no maxPendingMessages" presumption an invariant

With the flag set this returns before the maxPendingMessages semaphore is acquired, but completeCallbackAndReleaseSemaphore (:1468-1472) and releaseSemaphoreForSendOp (:1461-1465) still release it, so permits would climb past the configured maximum on every send.

Not reachable today: DEFAULT_MAX_PENDING_MESSAGES is 0 and the V5 builder exposes no setter, so the semaphore is absent. But the flag is a public Lombok setter on ProducerConfigurationData and the javadoc only "presumes" no limit. A checkArgument in the ProducerImpl constructor would make that presumption an invariant.