[fix][client] Release the command header when send serialization fails

[nodece]

Motivation

serializeCommandSendWithSize() and serializeCommandMessageWithSize() allocate the frame header and hand it, together with the payload, to the ByteBufPair only at the very end. If anything in between throws (e.g. an OOM while serializing the command or metadata), the header is orphaned and leaks.

Modifications

• Build the header inside a try block and create the ByteBufPair last. On any failure the header is released instead of being orphaned; the payload is deliberately not touched — releasing it on a failed send remains a pre-existing gap on the caller side (fixed for the producer send path separately).

Verifying this change

• Make sure that the change passes the CI checks.

This change is a small exception-safety fix; the serialized wire format is unchanged. Verified locally with CommandsTest, ByteBufPairTest, ProducerImplTest, ServerCnxTest and checkstyle.

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

• Dependencies (add or upgrade a dependency): no
• The public API: no
• The schema: no
• The default values of configurations: no
• The threading model: no
• The binary protocol: no — the wire bytes are unchanged
• The REST endpoints: no
• The admin CLI options: no
• The metrics: no
• Anything that affects deployment: no

Documentation

• no-doc needed (internal buffer-lifecycle fix, no user-facing behavior change)

[lhotari]

Thanks for picking this up — the buffer lifetimes around ByteBufPair are easy to get wrong, and the Commands half of this is a real fix: before it, a throw between the header allocation and the checksum block orphaned the header and the ByteBufPair already built around it. I confirmed the reordering of ByteBufPair.get(...) past the checksum computation is semantically inert.

The ByteBufPair.Encoder / CopyingEncoder half I'd ask you to reconsider. netty 4.2.17.Final — the version this branch builds against — has two different failure modes for ctx.write(...), and the new finally is correct for one and a double release for the other:

• a promise-validation failure releases the message and then rethrows (validateWrite), so the cleanup here releases it a second time;
• an Error from pipeline.touch(...) or WriteTask.newInstance(...), which run outside any cleanup block, throws without releasing — that is the (narrow) window the change genuinely closes.

The handler can't tell them apart from outside Netty. Taking the first case with the pair retained for more than one write — which the comment above this code explicitly supports — the extra release drops the payload below the refcount the outstanding ByteBufPair reference still needs: at this commit the pair is left at refCnt == 1 with its second buffer already at 0, where master leaves both at 1. Commands, output and the mechanism are in the comment on that block.

The case the new comment actually names — "a write rejected while the event loop is shutting down" — reaches neither branch: safeExecute(...) releases the message, fails the promise and returns without throwing, so the finally runs with both locals already null.

Two smaller points: the Commands change covers only the header, and the payload reference the ByteBufPair would have taken ownership of is still dropped by nobody; and testSerializeCommandSendFailureKeepsPayloadWithCaller passes unchanged with the Commands.java change reverted, so it isn't pinning the fix it was written for.

[lhotari]

[INTENT MISMATCH] no caller drops the payload reference the ByteBufPair would have owned, so the payload side of the leak remains

Releasing the header here is a real improvement. But the contract this comment leans on isn't there yet in either production caller, so the payload side is still dropped by nobody.

Client. ProducerImpl.serializeAndSendMessage → sendMessage(...) → Commands.newSend(...); when that throws the unwind is:

pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java

Lines 708 to 714 in e7aaa60

	} catch (PulsarClientException e) {
	e.setSequenceId(msg.getSequenceId());
	completeCallbackAndReleaseSemaphore(uncompressedSize, callback, e);
	} catch (Throwable t) {
	completeCallbackAndReleaseSemaphore(uncompressedSize, callback,
	new PulsarClientException(t, msg.getSequenceId()));
	}

completeCallbackAndReleaseSemaphore (ProducerImpl.java:1446-1450) releases the send semaphore and the memory-limit accounting, and the callback it invokes does release one reference — msg.getDataBuffer() at ProducerImpl.java:475-501, balancing the retain in internalSendAsync. That is not the reference the ByteBufPair was about to take ownership of. The lifecycle note on sendAsync itself spells out that the single-message path expects two releases, the second being ByteBufPair.release() (ProducerImpl.java:537-562) — and with no pair ever created that one never happens. With compression or encryption enabled it is a whole buffer rather than a reference: applyCompression (ProducerImpl.java:531-535) and encryptMessage release their source and return a new buffer whose only owner is the local variable.

Broker. PulsarCommandSenderImpl does metadataAndPayload.retain() at PulsarCommandSenderImpl.java:271 and only reaches entriesToRelease.add(entry) after ctx.write(...) returns (PulsarCommandSenderImpl.java:297-302), so a throw out of Commands.serializeCommandMessageWithSize skips both that and the flush listener that would have released the entries.

Failure scenario. A MessageMetadata whose getSerializedSize() disagrees with what writeTo() writes — a field mutated between the two calls, or a lazily-decoded field whose backing buffer is already gone — throws inside the new try. The header is now released; the payload reference is not, and a producer that hits this repeatedly walks its direct-memory pool down.

Either resolution is fine by me as long as the comment and the PR description match it: extend the fix to the callers, or drop "whose contract is to release it on failure" and say plainly that the payload side remains a pre-existing gap. Releasing the payload inside the serializer would not be right — op.rePopulate re-serializes with the same buffer, and the un-compressed batch path keeps its buffer for a retry.

[nodece]

Agreed, and addressed. The comment in serializeCommandSendWithSize / serializeCommandMessageWithSize no longer claims a payload-release contract — it now says plainly that the payload is deliberately not touched on failure and that releasing it on a failed send remains a pre-existing gap on the caller side.\n\nFor completeness: the payload side of the send failure paths is handled by the separate PR #26455 (sendMessageOrReleasePayload releases the payload when command serialization fails; schema-deferred ops hold the payload in pendingPayload and recycle() releases it; the batch paths use an ownership flag). Those two PRs are complementary: #26492 releases the header, #26455 releases the payload, and since the pair is created last, neither double-releases.

[lhotari]

[QUALITY] the mock covers only the ownership state in which the new code is correct, so it cannot detect the double release

when(ctx.write(any(), any())).thenThrow(...) throws without releasing the message. That is one of the two ownership states netty 4.2.17 can leave behind (an Error inside pipeline.touch(...) / WriteTask.newInstance(...)), and it is the one where the new code behaves correctly — so this test can't detect the other one, where Netty releases the message before rethrowing and the new cleanup releases it a second time. The pre-existing testEncoder in this same file already models that second state:

pulsar/pulsar-common/src/test/java/org/apache/pulsar/common/protocol/ByteBufPairTest.java

Lines 94 to 99 in e7aaa60

	ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);
	when(ctx.write(any(), any())).then(invocation -> {
	// Simulate a write on the context which releases the buffer
	((ByteBuf) invocation.getArguments()[0]).release();
	return null;
	});

A companion case that keeps thenThrow for the allocation-failure state and adds a release-then-throw case would cover both, and the second one is what surfaces the problem I described on the encoder itself.

[nodece]

Agreed — the companion release-then-throw case is exactly what surfaced the double release, so the whole encoder test (and the encoder change) was dropped rather than patched around.

[nodece]

Thanks @lhotari for the detailed review — all three points verified and addressed:

1. Encoder / CopyingEncoder change dropped. I reproduced the double release you described (release-then-throw write failure on a pair retained twice: pair.refCnt=1, b2.refCnt=0 at the previous commit vs b2.refCnt=1 on master) and reverted the encoder half entirely — ByteBufPair.java is now byte-identical to master.
2. Comment corrected. serializeCommandSendWithSize / serializeCommandMessageWithSize still build the header inside the try and create the pair last, but the comment now states plainly that the payload is deliberately not touched on failure and that releasing it on a failed send remains a pre-existing gap on the caller side (the producer-side release is handled by the separate send-path fix).
3. Ineffective test removed. testSerializeCommandSendFailureKeepsPayloadWithCaller (which passed with the Commands.java change reverted) and testEncoderReleasesDuplicatesWhenWriteFails are gone; ByteBufPairTest is back to the master test set. A seam-based test for the header release would require exposing the header allocation, which is more churn than this one-hunk fix warrants; the wire-format is covered by the existing CommandsTest.

The PR is now a single-commit, single-file change (only Commands.java).