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

[nodece]

Motivation

When a send fails midway, every buffer handed through the pipeline must be released by the component that last took its ownership. This PR fixes the paths that violated this and leaked buffers instead — leaks that accumulate exactly under the sustained-failure conditions (e.g. direct-memory pressure) that trigger the failures in the first place.

Batch flush failure recovery (original scope): when compression or encryption releases the batch buffer before a flush that then fails, resetPayloadAfterFailedPublishing() could not tell whether the container still owned the buffer: retrying reused the released buffer (use-after-free), and the compressed/encrypted payload that replaced it was left orphaned (leak).

Multi-batch partial build (from review, @void-ptr974): when a later sub-batch fails to build, the operations already built never reached the send queue and their commands leaked.

Chunked payload base claim: 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. A serialization failure on a non-last chunk orphaned that claim, leaking the shared payload buffer on every failed chunked send (found while building the wiring tests for this PR).

Encrypted payload on the send failure paths: the payload buffer produced by the encryption step leaked at all three failure points of its lifecycle — during encryption (the partially built output buffer), after encryption (a failed command serialization orphaned it; for chunked messages the retained slice also leaked its claim on the shared base buffer), and when the command was deferred until schema registration and the op was failed first (send timeout, producer close) — the payload leaked with the unreachable rePopulate closure.

Modifications

• BatchMessageContainerImpl: track batch-buffer ownership with an explicit batchPayloadOwned flag instead of inferring it from the buffer reference. A build only starts while the container owns its buffer; compression/encryption that releases it clears the flag, and resetPayloadAfterFailedPublishing() reallocates instead of reusing released memory. The compressed payload is released when encryption fails and the container no longer owns it (fixes the orphaned-payload leak). resetPayloadAfterFailedPublishing() also skips reallocation when the container has no messages left, since key-based batching forwards the reset to sub-batches that already succeeded and cleared.
• RawBatchMessageContainerImpl: symmetrically release the compressed payload and any partially built encrypted buffer when encryption fails, and release the serialized payloads in toByteBuf() on failure instead of orphaning them.
• StrategicTwoPhaseCompactor: always clear the batch container on a failed flush, including when the failure is an Error (discard(Exception) cannot take an Error).
• Multi-batch build (BatchMessageKeyBasedContainer, EntryBucketBatchContainer): build sub-batches in an explicit loop; when a later sub-batch fails, release the commands already built (ownership-aware: the header is always solely owned by the command; the payload only when its ownership left the container). Messages stay in their sub-batches and are retried, so their semaphore permits and memory reservations settle when the retry completes.
• ProducerImpl:
  • 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.
  • The chunk loop in sendAsync() releases the base payload's own ref-count claim when a non-last chunk's serialization fails; the last chunk is excluded because its unretained slice already is that claim, released by the serialization-stage helper.
• Tests: failure recovery with/without compression, encryption-failure buffer release, the fail-fast guard against re-entering a build without reset, multi-batch partial-build failure (leaked command, with and without compression), and the three encrypted-payload failure points.

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows: BatchMessageContainerImplTest (failure recovery, multi-batch partial build, entry-bucket stamping), RawBatchMessageContainerImplTest (encryption-failure releases), ProducerImplTest (partial encrypted-buffer release on crypto failure — rethrow and SEND fallback, payload release on failed command serialization, deferred payload release on recycle).

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

• Dependencies (does it pull in a new dependency?): no
• The public API: no
• The schema registry: no
• The default values of configurations: no
• The wire protocol: no
• The rest endpoints: no
• The admin cli options: no
• Anything that affects deployment: no

Documentation

• no-doc needed (internal leak fixes, no user-facing behavior or configuration change)

[void-ptr974]

One additional case worth considering is partial success during multi-batch construction. For example, sub-batch A successfully creates an OpSendMsg, but sub-batch B fails while constructing its command. In this situation, A’s operation becomes unreachable and its ByteBufPair is not released.

Repeated retries may retain additional command buffers and cause direct-memory growth or eventually OOM. With CompressionType.NONE, the abandoned command may also reference the payload reused during retry, resulting in inconsistent buffer ownership.

[nodece]

@void-ptr974 Good catch — confirmed. createOpSendMsgs() built the sub-batches through a stream, so a mid-loop failure dropped the already-built operations and leaked their commands.

Two notes from digging in:

1. The messages are not lost — createOpSendMsg() doesn't consume them, so they stay in their sub-batches and are retried on the next flush; the damage is the leaked command buffers (plus their memory reservations never settling).
2. The cleanup has to be ownership-aware: the payload is only released when compression/encryption handed it to the command; without compression the container still owns the buffer and the retry reuses it.

Fixed in 6eb6971 for both the key-based and the entry-bucket containers, with a regression test that fails on the previous code (leaked header buffer).

[void-ptr974]

One remaining case is an encryption failure in the non-batch path. encryptMessage() leaves the source payload to its caller, but the caller does not release it when the exception propagates. Repeated failures can therefore leak direct memory. A caller-path cleanup and regression test would cover this case.

[void-ptr974]

One minor lifecycle detail remains: releaseOrphanedOpCmd() releases the internal buffers, but does not release the ByteBufPair or recycle the OpSendMsg. Repeated failures may therefore add unnecessary heap allocation and GC pressure.

[nodece]

@void-ptr974 Right — fixed in 6fc4bf2. One detail kept it from being a plain op.cmd.release(): without compression or encryption the payload claim is shared between the pair and the container (single refcount), so releasing the pair directly would free the buffer the container still reuses for the retry. The fix takes the container's claim out of the pair with a retain() first, then releases the pair itself (back to its recycler) and recycles the op. The regression test now also asserts the pair's refCnt reaches 0.

[nodece]

@void-ptr974 Right — the caller-side cleanup was missing. Fixed in 9eed0a9: new applyCompressionOrReleaseSource() / encryptMessageOrReleaseSource() wrap both stages, releasing the source payload (and the chunk slice's claim on the shared base buffer) when the stage throws, mirroring sendMessageOrReleasePayload() on the serialization stage. Applied at all three call sites that lacked cleanup — the two in serializeAndSendMessage() and the pre-chunking compression in the send loop. Both helpers have regression tests.

[void-ptr974]

Thanks for adding the deferred-payload cleanup. There is one retry case worth covering here. For example, a multi-schema producer sends the first message with a new schema, schema registration succeeds, and Commands.newSend fails once while allocating the SEND header under direct-memory pressure. rePopulate clears pendingPayload before sendMessageOrReleasePayload runs, so the failure releases the payload, but the op remains in pendingMessages with cmd == null. If the producer later reconnects and resends its pending operations, the same closure uses the released buffer again, which can cause an IllegalReferenceCountException and leave the send stuck. Keeping pendingPayload until command construction succeeds, or removing and failing the op when releasing it, together with a test covering a failed first construction followed by recovery, would handle this case.

[nodece]

@void-ptr974 Good catch — the release-on-throw semantics were wrong for this path: the op itself is the retry owner of the deferred payload. Fixed in 78768d1: the closure body is now buildDeferredCommand(), which builds from op.pendingPayload and clears the field only after the construction succeeds — a failed construction keeps the payload with the op for the next resend, and recycle() still releases it when the op is failed instead. Added the requested test: a failed first construction followed by recovery, asserting the payload survives the failure, the retry rebuilds the command from the same buffer, and a later recycle does not double-release.

[void-ptr974]

LGTM

[lhotari]

The buffer-ownership fixes look sound, including cleanup of partially built multi-batch operations and retaining the deferred payload until command construction succeeds. I found no new production regression. The remaining comments concern regression-test coverage, the SEND-frame parsing offset, and the new private-state reflection in tests.

[nodece]

@lhotari All four points addressed, plus one more leak the wiring work surfaced — in 5031cf2 and e47b27f.

Send-path wiring — testSendPathFailureReleasesPayloadThroughTheStageHelpers drives the real sendAsync(): a failing compression stage and a failing command serialization both release the payload, and reverting either call site to the bare method fails the test (verified by running the reversions). testDeferredSchemaOpWiringThroughSendPath additionally drives the real deferred branch: it captures the op and asserts the pendingPayload assignment, the failed-first-build retention, the resend rebuild, and the recycle release. Two chunked variants (testChunkedSendFailure*) cover the chunk-loop call sites.

RawBatch encrypted-output release — the container gained a @VisibleForTesting constructor taking the allocator; both encryption-failure tests now track every buffer the container allocates and assert all of them are freed, so removing the encrypted-output release fails them (verified). The PulsarClientException branch gets the same tracked coverage, asserting the batch is discarded for container reuse.

skipBytes(8) — fixed, with a comment noting both length fields.

Reflection — the producers in these tests are now built through their real constructors (useConstructor + CALLS_REAL_METHODS): conf, log, client and msgCrypto hold real values, and the crypto mock is installed via conf.setMessageCrypto() through the same constructor branch production uses. No new reflection; producer stubs use the doAnswer form throughout, since when-form registration executes the real method on these mocks.

New leak found while building the wiring tests: a failing non-last chunk of a chunked persistent message orphaned the base payload's own ref-count claim — the last chunk carries it (its slice is never retained), so a mid-loop failure leaked the shared buffer on every failed chunked send. The chunk loop now releases it (e47b27f), with the two chunked tests failing on the reversion.