[Denovo1998]

Motivation

PulsarCommandSenderImpl retains a large message's payload before building its command. If serialization fails before a ByteBufPair is returned, dispatch cleanup closes the connection and releases the entry, but leaves the extra payload reference behind. The buffer can remain allocated after the dispatch has failed.

This ownership gap predates #26926. The header cleanup in #26492 does not release the caller's retained payload reference.

Modifications

• Construct the message before retaining its payload, so a failed serialization leaves no extra payload reference to release.
• Retain the payload for the channel only after construction succeeds, preserving Netty's ownership after handoff, including failed writes.
• Extend PulsarCommandSenderImplTest with failures on the first large message and midway through a mixed dispatch, plus successful-write and closed-channel controls.

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• Before the fix, both allocation-failure cases leave the failed payload at refCnt == 1 after the header and entry are released. The other four sender cases pass.
• With the fix, PulsarCommandSenderImplTest passes all 6 cases and DispatchSmallMessagesCoalescingTest passes all 3 cases, with zero failures or skips and retries disabled.
• quickCheck passes across the repository.

./gradlew :pulsar-broker:test \
  --tests 'org.apache.pulsar.broker.service.PulsarCommandSenderImplTest' \
  --tests 'org.apache.pulsar.broker.service.DispatchSmallMessagesCoalescingTest' \
  -PtestRetryCount=0 -PtestFailFast=false -PtestMaxParallelForks=1 --max-workers=1
./gradlew quickCheck --max-workers=1

The failure tests inject an OutOfMemoryError only at ByteBufPair.get, since the static allocator/recycler has no local failure hook. The sender, ServerCnx, serializer, entries, buffers, and Netty write handling use their real implementations. The tests do not exhaust JVM memory or validate end-to-end reconnect/redelivery after this failure.

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

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment

[merlimat]

Thanks. The ownership gap is real: the pair only takes over the payload reference once ByteBufPair.get returns, and the dispatch catch block only releases the entry's own reference. The test reproduces it for the right reason. One optional simplification inline.

[codelipenghui]

Reviewed 10c49ce. No actionable findings identified within the reviewed scope.

The construction-only cleanup balances the retained payload while preserving Netty's ownership after write handoff. All 6 PulsarCommandSenderImplTest cases and 3 DispatchSmallMessagesCoalescingTest cases passed, with no failures or skips and retries disabled. Temporarily removing the cleanup reproduced the first-message payload leak (refCnt 1 instead of 0); fail-fast skipped the midpoint baseline case. The exact PR code was restored afterward.

Validation used injected allocation failure, not real memory exhaustion or end-to-end reconnect/redelivery.