[fix][client] Fix buffer ownership on the send failure paths - #26455

…atch fails

Motivation

createOpSendMsgs() built all sub-batches through a stream that collected
into a list. When a later sub-batch failed to build, the stream aborted
and the operations already built were unreachable: they never reached the
send queue, so nothing released their commands. Every failed flush then
leaked the command buffers (serialized header plus the transferred batch
payload), and repeated retries under a persistent failure cause would
keep growing the direct memory usage.

Modifications

- BatchMessageKeyBasedContainer and EntryBucketBatchContainer build their
  sub-batches in an explicit loop; on failure they release the commands
  of the operations already built before rethrowing.
- The release is ownership-aware (releaseOrphanedOpCmd): the serialized
  header is always solely owned by the command, while the payload is only
  released when its ownership left the container (compression or
  encryption); otherwise the container keeps the buffer and the retry
  after resetPayloadAfterFailedPublishing() reuses it.
- The messages stay in their sub-batches, so their semaphore permits and
  memory reservations settle when the retry completes.
- Add a regression test for the partial multi-batch failure with and
  without compression (fails on the previous code with a leaked header).

Motivation

The payload buffer produced by the encryption step must be released when
a send fails midway - by whichever component last took its ownership.
All three failure points along its lifecycle leaked it instead, so every
failed send leaked direct memory, accumulating exactly under the
sustained-failure conditions (e.g. direct-memory pressure) that trigger
these failures in the first place:

1. During encryption: encryptMessage() leaked the partially built output
   buffer whenever the crypto failed - on the PulsarClientException
   rethrow, on the SEND (publish-unencrypted) fallback, and on any
   RuntimeException or Error escaping the crypto, which had no cleanup
   at all. The batch container path only released the source payload
   around this call (previous commit); the output buffer allocated
   inside was unreachable to it.
2. After encryption: serializeAndSendMessage() orphaned the encrypted
   payload when building the send command threw (e.g. a header
   allocation failure). For chunked messages the retained slice also
   leaked its claim on the shared base buffer, so the base never
   returned to the pool.
3. Deferred command: an operation whose command is built only after the
   schema registration completes (rePopulate) captured the encrypted
   payload in a closure; if the op was failed first (send timeout,
   producer close), the payload leaked with the unreachable closure.

This completes the failure-path ownership fixes for the non-batch send
path; the batch flush path is covered by the previous commits.

Modifications

- encryptMessage() releases the partially built output buffer on every
  exit that does not hand it to the caller; the source payload remains
  the caller's responsibility (unchanged contract).
- New sendMessageOrReleasePayload() used at both command-building sites
  in serializeAndSendMessage(): a failed serialization releases the
  encrypted payload (and the chunk slice's claim on the base buffer)
  before rethrowing.
- OpSendMsg tracks the deferred payload (pendingPayload); rePopulate()
  clears it once the payload moved into the command, and recycle()
  releases it when the op is failed before that.
- Tests for all three failure points.

…eanup

Motivation

releaseOrphanedOpCmd() freed the pair's component buffers but left the
ByteBufPair and the OpSendMsg unreleased: both are recycler-based, so on
repeated multi-batch build failures they were garbage-collected instead
of returned to their pools, adding heap allocation and GC pressure (noted
in review).

Modifications

- When the payload claim is shared with the container (no compression or
  encryption), take it out of the pair with a retain() before releasing,
  so resetPayloadAfterFailedPublishing() keeps reusing the buffer.
- Release the pair itself (returning it to its recycler) and recycle the
  orphaned op.
- The regression test now also asserts the pair itself is released.

…on fails

Motivation

Both applyCompression() and encryptMessage() leave their source payload with
the caller on failure (they release it only after success), but the non-batch
send path had no caller-side cleanup: when either stage threw, the exception
propagated out of serializeAndSendMessage() and the source payload was
orphaned. For chunked messages the retained slice's claim on the shared base
buffer leaked as well, so the base never returned to the pool. Repeated
failures therefore leaked direct memory (noted in review).

The batch-container path already defends around these calls via its ownership
tracking; the non-batch path did not.

Modifications

- New applyCompressionOrReleaseSource() and encryptMessageOrReleaseSource()
  used in serializeAndSendMessage(): on failure they release the source
  payload (and the chunk slice's claim) before rethrowing, mirroring
  sendMessageOrReleasePayload() on the serialization stage.
- Tests for both stages.

…on fails

Motivation

A deferred command (schema registration pending) released its payload when
the construction failed: rePopulate cleared pendingPayload before calling
the releasing serialization wrapper. The op stays in pendingMessages after
such a failure, and the next resend invokes the closure again - on the
released buffer, producing an IllegalReferenceCountException on every
reconnect and leaving the send stuck (noted in review).

The release-on-throw semantics were wrong for this path: unlike the
non-deferred send path, the op itself is the retry owner of the payload.

Modifications

- Extract the closure body into buildDeferredCommand(): the command is
  built from op.pendingPayload and the field is cleared only after the
  construction succeeds. A failed construction keeps the payload with the
  op for the next resend; recycle() still releases it when the op is
  failed instead.
- Regression test: a failed first construction followed by recovery -
  the payload survives the failure, the retry rebuilds the command from
  the same buffer, and recycling the op afterwards does not double-release.

…ld fails

Motivation

For chunked persistent messages, the non-last chunks retain their slices
while the base payload's own ref-count claim is carried by the last chunk
(its slice is never retained). When serialization failed on a non-last
chunk, the send-path helper released that chunk's retained-slice claim,
but the base's own claim was orphaned: every failed chunked send leaked
the shared payload buffer.

Modifications

- The chunk loop in sendAsync() releases the base payload's own claim
  when a non-last chunk's serialization fails. The last chunk is
  excluded because its unretained slice already is the base claim, which
  the send-path helper releases; the persistence condition mirrors the
  slicing condition in serializeAndSendMessage().
- Wiring-level regression tests: chunked failures with and without
  compression assert the base is fully released (reverting the release
  fails them, verified); a deferred-schema test drives the real
  sendAsync() into the deferred branch and asserts the pendingPayload
  assignment, the failed-first-build retention, the resend rebuild, and
  the recycle release; the RawBatch PulsarClientException branch and the
  entry-bucket partial-build loop get the same tracked-buffer coverage.

Two review findings on apache#26455, both about accounting acquired before an
OpSendMsg exists to release it through the send lifecycle:

- Non-batch compression failure: canEnqueueRequest has already acquired
  the send permit and reserved the message size when the compression
  stage throws, and no op is ever created, so nothing released them.
  The exception propagated out of sendAsync() synchronously, leaving
  the returned future incomplete, the permit consumed, and the memory
  reserved - repeated compression failures exhausted the producer queue
  and the client memory limit. The failure now completes the callback
  through completeCallbackAndReleaseSemaphore, releasing the accounting
  and failing the future with the cause.

- Chunked non-blocking failure: the chunk loop pre-acquires a permit
  for every chunk of the message before building any of them. A
  chunk-build failure released only the failing chunk's permit (via the
  outer catch), stranding the totalChunks - chunkId - 1 permits of the
  chunks that were never built - failing the first of 11 chunks with a
  queue of 100 left 90 - and their claims on the ChunkedMessageCtx,
  which only created ops release. The failure handling now returns the
  never-built chunks' permits (non-blocking mode only; blocking mode
  acquires per chunk inside the loop) and releases the corresponding
  context claims.

Regression tests drive the real sendAsync(): both failures must leave
the memory usage at zero and the available permits back at their
starting point. The tests set maxPendingMessages explicitly because the
configuration default of 0 means an unbounded queue with no permit
accounting. Verified against the previous code: the chunked test fails
with "expected [100] but found [90]", and the compression test sees the
raw RuntimeException escape sendAsync() with the future incomplete.

The existing send-path wiring test is updated to the new contract: a
failing compression stage completes the callback instead of throwing
synchronously.

Assisted-by: Claude Code

Tighten the failure-path comments to the invariant they protect and drop
the lines that repeat what the next statement already says; the
ownership reasoning itself stays.

…t chunk op only

The client memory for a chunked message is reserved once, by
canEnqueueRequest before the chunk loop, but every release site
(per-op ack, failPendingMessages, terminal state, recoverNotAllowedError)
releases op.uncompressedSize - and every chunk op carried the FULL
message size, since OpSendMsg.create reads it from the shared
MessageImpl. Each of the N chunk acks therefore released the whole
reservation again, and the memory gauge drifted low by (N-1) sizes,
silently disabling the client memory limit for chunked producers.

Only the last chunk's op now carries the size, so across all the
release paths the reservation is released exactly once: the last
chunk's ack in the success path, the last chunk op in the pending
queue in the failure paths, and the outer catch of sendAsync when the
last chunk was never built (the failure paths of a partial chunk loop
release 0 for the never-completed remainder).

Pre-existing, surfaced by the queue-accounting audit in apache#26455.
Regression test drives the real sendAsync chunk loop and asserts the
ops of one message carry the reservation exactly once in total
(fails with "chunk 0 of 11 ... expected [0] but found [10240]" on the
previous code).

Assisted-by: Claude Code

…lback throws

Review finding on apache#26455: discard() invoked the application callback
and released the batch buffer inside the same try block. The callback
runs application future handlers, so it can throw; the escaping
exception skipped the release, and the clear() that followed dropped
the buffer's reference without freeing it - every failed batch with a
throwing handler leaked its off-heap memory. The release, the
ownership flag reset, and clear() now run in a finally, with the
callback exception still caught and logged in isolation.

The try shape predates this PR (upstream has the same callback-then-
release ordering, without the ownership flag); the reworked discard
inherited it.

Regression test: a throwing sendComplete must leave the batch buffer
released and the container cleared (fails with "expected [0] but found
[1]" on the previous code).

Assisted-by: Claude Code

…e failure

Review finding on apache#26455: the blocking-mode chunk failure branch (per-chunk
canEnqueueRequest failing, which in blocking mode only happens on
interruption) released uncompressedSize - readStartIndex of the memory and
totalChunks - chunkId permits. Both are now wrong since 07dab10 ("Carry
a chunked message's memory reservation on the last chunk op only"): the
earlier chunk ops carry no memory share, so the already-chunked readStartIndex
bytes were never released and the gauge crept up on every failure - a leak.
The permit release was also an over-release all along: in blocking mode the
chunks after the failing one were never acquired (each chunk takes its permit
inside the loop), and the failing chunk's own attempt aborts before acquiring.

The branch now releases the whole memory reservation and no permits; the
already-built chunks' permits are returned by their own operations.

Also:
- completeCallbackAndReleaseSemaphore documents why its unconditional permit
  release is safe: the constructor forbids combining
  memoryLimitAdmittedUpstream with a maxPendingMessages limit, so a producer
  with admission upstream has no semaphore to release from.
- RawBatchMessageContainerImpl.encrypt's Throwable-branch comment now states
  precisely why the batch is left to the caller's discard: the failure escapes
  toByteBuf() before its serialization try (whose finally clears the
  container).

Regression test interrupts the thread between chunk builds
(Semaphore.acquire() throws immediately on an interrupt flag) and asserts
the memory gauge and the permit count return to their starting points
(fails with "expected [0] but found [1005]" on the previous code).

Assisted-by: Claude Code

…ls it

Two review findings on apache#26455, both queue/buffer paths left stranded by
a failure:

- processOpSendMsg's catch (from apache#23738) released only the permit and
  the memory when the body threw - realistically when the connection's
  event loop rejected the write task while shutting down. The op stayed
  in pendingMessages unrecycled, its cmd leaked both references (the
  op's own and the one retained for a write that never got queued), and
  a chunked op's ChunkedMessageCtx claim leaked with it. The catch now
  takes the op back out of the queue (OpSendMsgQueue gained a
  remove(OpSendMsg) that keeps the message-count accounting; the
  iterator's remove already did), drops the orphaned write reference,
  and releases the op's own command via releaseOpCmdAndRecycle, whose
  inline fallback covers the rejecting event loop.

- In the resend recovery loop, a rebuilt command that exceeds the max
  message size took the isMessageSizeExceeded path, which already
  released the accounting and completed the callback - but the bare
  continue left the op in pendingMessages with its fresh cmd, so it sat
  until the send timeout failed it and released the accounting a second
  time. The continue now removes the op from the iteration and releases
  its command and the op itself.

Regression test drives the real processOpSendMsg with an event loop
that rejects every task and asserts the queue is empty, the accounting
follows, and the cmd reaches refCnt 0 (fails with "expected [0] but
found [1]" queue residue on the previous code).

Assisted-by: Claude Code

The catch's op.sendComplete() was the only unprotected call site: the
callback runs application future handlers and can throw, and by the time
it runs the op is already out of pendingMessages with its accounting
released - an escaping exception skipped the cmd references, the
chunked-message context claim, and the op recycling for good. Wrap it
like every other sendComplete site (failPendingMessages, ackReceived,
recoverNotAllowedError), keeping the releases after it unconditional.

Regression test combines the rejecting event loop with a throwing
callback and asserts the queue is empty and the cmd reaches refCnt 0
(the previous code fails with the callback's RuntimeException escaping).

Assisted-by: Claude Code

…use-after-release