[fix][client] Fix two producer buffer-ownership defects: oversized batches and non-persistent chunking

[SongOf]

Motivation

Two independent buffer-ownership defects on the producer send path. They are unrelated in cause, so they
are kept as separate commits, but both end in the same place: a pooled buffer released more often than it
was retained, which hands a live buffer back to the allocator.

1. An oversized batch releases its payload twice.

BatchMessageContainerImpl holds the batch payload in batchedMessageMetadataAndPayload. The ownership
rule is set by the success path: once the payload is handed to a ByteBufPair, that pair owns it, and
clear() only nulls the field instead of releasing it.

Both oversized branches of createOpSendMsg break that rule by letting two owners release the same buffer.
With compression and encryption off — the default — getCompressedBatchMetadataAndPayload() returns the
field itself and encryptMessage passes it straight through, so all three references are the same buffer:

// single-message branch
if (op.getMessageHeaderAndPayloadSize() > getMaxMessageSize()) {
    cmd.release();       // correct: the command owns the payload now
    ...
    discard(...);        // releases batchedMessageMetadataAndPayload — the same buffer, a second time

// multi-message branch: no command exists yet, so the container still owns the payload
if (encryptedPayload.readableBytes() > getMaxMessageSize()) {
    encryptedPayload.release();
    ...
    discard(...);        // releases it again

This is reached by the ordinary path: batching on (the default) and a message over the maximum message
size. If the buffer has already been handed out again by the pool, the second release frees memory that is
still in use.

The same file has a related dangling reference. applyCompression releases the buffer it is given and
returns a new one, but the field was never updated to follow it, so from that point on it pointed at freed
memory that discard() would release again and resetPayloadAfterFailedPublishing() would write into. The
non-client branch of getCompressedBatchMetadataAndPayload(boolean) has the same shape and is reached by
RawBatchMessageContainerImpl, which sets the compression type from the message being re-batched.

2. Messages are "chunked" on non-persistent topics.

Chunking has been guarded against non-persistent topics since it was introduced (PIP-37, #4400), but only
at the point where chunking is performed, not where it is decided:

// how many chunks — no topic-type check
totalChunks = MathUtils.ceilDiv(Math.max(1, compressedPayload.readableBytes()), payloadChunkSize);

// performing the chunking — guarded
if (totalChunks > 1 && TopicName.get(topic).isPersistent()) {
    chunkPayload = compressedPayload.slice(readStartIndex, ...);
    if (chunkId != totalChunks - 1) {
        chunkPayload.retain();
    }
    msgMetadata.setChunkId(chunkId).setNumChunksFromMsg(totalChunks)...;
}

The send loop runs totalChunks times regardless. On a non-persistent topic the guarded block is skipped
while the loop still runs N times, so three things follow from that one skip:

• chunkPayload stays the whole payload, which is published N times;
• no chunk metadata is written, so the consumer cannot tell these are one message and delivers N duplicates
  that deduplication cannot catch either;
• the compensating retain() lives inside the skipped block, so the payload is handed to N ByteBufPairs
  while only one reference is held — N releases against a refcount of 1.

Nothing rejects the configuration: the builder only refuses chunking together with batching.

Modifications

Oversized batch release. Give each buffer exactly one owner, and drop the stale reference whenever
ownership moves. The single-message branch keeps cmd.release() — the command does own the payload by then
— and nulls the field before discard() runs, mirroring what clear() does on the success path. The
multi-message branch releases encryptedPayload only when it really is a different buffer, which is the
case exactly when encryption produced one; otherwise discard() is the single owner. Both compression
branches now assign the new buffer back to the field.

Chunking on non-persistent topics. Move the topic-type check from where chunking is performed to where
it is decided, so the two agree: a non-persistent topic computes totalChunks = 1 and takes the ordinary
single-message path. The check at the slicing site then becomes redundant and is reduced to totalChunks > 1, matching the sibling site in the same method that has always tested only the chunk count. The repeated
TopicName.get(topic) parse on the send path is replaced by a field computed once in the constructor.

Because the duplicate publishing and the refcount underflow are two symptoms of the same skipped block,
unifying the condition removes both; no separate accounting change is needed.

No wire format or API change. Applications sending large messages to a non-persistent topic with chunking
enabled now publish them once instead of N times; the message must still fit the maximum message size, as
it did before for every one of those N copies.

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• Added BatchMessageContainerImplTest.testOversizedBatchReleasesItsPayloadExactlyOnce: a recording
  allocator hands out the batch buffer so the test can hold it without reflecting into private state; the
  buffer is retained once to stand in for another holder, so an extra release is observable instead of
  being swallowed by ReferenceCountUtil.safeRelease. It fails before the change with a reference count
  of 0 where 1 is expected.
• Added MessageChunkingTest.testLargeMessageOnNonPersistentTopicIsSentOnceWithoutChunking: a payload
  larger than chunkMaxMessageSize is sent to a non-persistent topic and must be delivered exactly once.
  It fails before the change with "the payload was published more than once on a non-persistent topic".
• BatchMessageContainerImplTest (5), MessageChunkingTest (18) and RawBatchMessageContainerImplTest
  (7) pass; the pulsar-client module suite shows no new failures and ./gradlew quickCheck is clean.

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
• 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

Chunking is now inert on non-persistent topics, where it never produced a message a consumer could
reassemble. The setting is still accepted.

[lhotari]

Thanks for the fixes and regression tests. The non-persistent chunk-count change looks consistent with the send loop, and the unencrypted batch ownership paths are improved. One oversized encrypted-batch path still needs the same ownership correction. I reviewed the code and tests but did not run the tests locally.

[lhotari]

[BUG] Encrypted oversized batches still release the input buffer twice

Could you also transfer the container field to the buffer returned by encryption and cover successful encryption in the regression test? With a multi-message batch whose encrypted payload exceeds the limit, encryptMessage has already released its input before returning a different buffer:

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

Lines 1012 to 1014 in 8352e29

	encryptedPayload.writerIndex(targetBuffer.remaining());
	compressedPayload.release();
	return encryptedPayload;

The new inequality branch releases the encrypted output, but discard() still releases batchedMessageMetadataAndPayload, which points to that already-released input. This leaves the oversized-batch double release in place when encryption succeeds, both with and without compression. The current test mocks encryption as an identity operation, so it cannot exercise this case.

[SongOf]

Confirmed and fixed in 72bc03e. You are right that the inequality branch only settled who releases the encrypted output, not the stale field: encryptMessage releases its input and returns a different buffer on success, so batchedMessageMetadataAndPayload pointed at freed memory that discard() released again.

The field now follows whatever encryption returns, at both call sites, exactly as it already does for compression — which is the line I should have extended one call further. That makes the inequality check redundant, so it is gone and the branch is simpler than before. All four exits of encryptMessage are covered: encryption disabled and the failure-with-SEND path return the input, so the assignment is a no-op; success transfers ownership to the new buffer; the failure-with-FAIL path throws before the assignment, leaving the field on the unreleased input for resetPayloadAfterFailedPublishing.

Added testOversizedEncryptedBatchReleasesItsBuffersExactlyOnce, which mocks encryption the way the real one behaves (release the input, return a new buffer). The load-bearing assertion is on the pre-encryption buffer, since the encrypted output was already released exactly once before this change. It fails without the fix with "the pre-encryption batch payload was released again after encryption had already released it".

BatchMessageContainerImplTest passes 6/6, the pulsar-client module suite (494 tests) shows no new failures, RawBatchMessageContainerImplTest (7) and MessageChunkingTest (18) pass, and quickCheck is clean.