[fix][client] Make chunking inert on non-persistent topics, including the message-size checks

[SongOf]

Motivation

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.

The message-size checks have the same blind spot. Both the compressed-size check in sendAsync() and
isMessageSizeExceeded() skip validation whenever conf.isChunkingEnabled() is true, on the assumption
that an oversized message will be chunked. On a non-persistent topic it never is, so a payload above the
broker limit is sent as a single oversized frame instead of being rejected locally with
InvalidMessageException.

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

Note: an earlier revision of this PR also fixed an oversized-batch double release in
BatchMessageContainerImpl. That defect has since been fixed on master by #26455 (buffer ownership on
the send failure paths), so that part was dropped when merging master. Regression tests for the oversized
branches are kept, since #26455 did not add one for that path.

Modifications

Introduce a single private ProducerImpl.isChunkingEnabled() that returns
conf.isChunkingEnabled() && persistentTopic, and route every decision that depends on chunking through it:

• the chunk computation, so a non-persistent topic computes totalChunks = 1 and takes the ordinary
  single-message path (the slicing site's own topic check then becomes redundant and is reduced to
  totalChunks > 1, matching the sibling site in the same method);
• the compressed-size check in sendAsync();
• isMessageSizeExceeded().

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. The repeated
TopicName.get(topic) parse on the send path is replaced by a field computed once in the constructor.

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, and a message above the broker limit fails locally with
InvalidMessageException exactly as it does with chunking disabled.

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• 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".
• Added MessageChunkingTest.testLargeMessageOnNonPersistentTopicAboveBrokerLimitIsRejectedLocally: a
  payload larger than the broker's maxMessageSize plus frame padding is sent to a non-persistent topic
  with chunking enabled and must be rejected locally with InvalidMessageException. Before the change
  the send is not rejected by the client.
• Added three regression tests in BatchMessageContainerImplTest for the oversized-batch branches now
  covered by [fix][client] Fix buffer ownership on the send failure paths #26455: testOversizedSingleMessageBatchReleasesItsPayloadExactlyOnce (single-message branch,
  where cmd.release() is the only legitimate release), testOversizedBatchReleasesItsPayloadExactlyOnce
  (multi-message branch, container-owned buffer) and testOversizedEncryptedBatchReleasesItsBuffersExactlyOnce
  (multi-message branch after a successful encryption). Each fails when the corresponding ownership handling
  is removed.
• MessageChunkingTest (19), BatchMessageContainerImplTest (25) and RawBatchMessageContainerImplTest
  (10) pass, 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.

[Denovo1998]

Can we update both message-size checks to apply the same effective chunking condition? For a non-persistent topic, totalChunks is currently set to 1, but both the payload-size check in sendAsync() and isMessageSizeExceeded() still skip validation when conf.isChunkingEnabled() returns true.

I reproduced this issue with a 1 KiB broker limit and a 32 KiB payload: the broker repeatedly reports TooLongFrameException, and the send eventually fails with a TimeoutException instead of a locally caught InvalidMessageException. A consistent check for both chunkingEnabled and persistentTopic should address this. Additionally, could we add a test above the broker limit? The current 500-byte test only exceeds the chunk size threshold.

[SongOf]

Fixed in 1ba0e84. Both size checks were still gated on conf.isChunkingEnabled() alone, so on a non-persistent topic the chunk count was 1 while the checks still assumed the payload would be chunked.

There is now a single private isChunkingEnabled() on ProducerImpl that returns conf.isChunkingEnabled() && persistentTopic, and all three sites go through it: the chunk computation, the compressed-size check in sendAsync(), and isMessageSizeExceeded().

Added testLargeMessageOnNonPersistentTopicAboveBrokerLimitIsRejectedLocally: a payload above the broker's maxMessageSize plus frame padding, sent to a non-persistent topic with chunking enabled, must fail locally with InvalidMessageException. Without the fix the test fails because the client does not reject the send. One detail differs from your repro: in the test harness the broker's frame decoder limit is fixed at startup, so the unfixed send completed instead of timing out, but the assertion is on the local rejection, so it discriminates either way.

[Denovo1998]

Should we also update RawBatchMessageContainerImpl.encrypt() during this ownership transfer? After the assignment, the container field points to compressedPayload, but the encryption-failure handler still invokes compressedPayload.release() and then discard(e), resulting in discard() releasing the same buffer again.

I reproduced this by injecting a CryptoException and keeping the compressed input retained once—its refcount ends at 0 instead of 1. Leaving the input release to discard() resolves that case.

[SongOf]

Confirmed on the branch as it was: after the field was transferred to the compressed buffer, the encryption-failure handler released it and discard() released it again.

After merging master this no longer applies. #26455 replaced the field-transfer approach with the batchPayloadOwned flag: compression clears the flag, discard() only releases the field when the flag is set, so the compressedPayload.release() in the failure handler is now the single release. #26455 also added testToByteBufReleasesPayloadAndDiscardsWhenEncryptionFailsWithClientException for exactly this path. This PR therefore no longer touches BatchMessageContainerImpl or RawBatchMessageContainerImpl; the diff against master is down to the chunking change in ProducerImpl plus tests. The PR title and description are updated accordingly.

[Denovo1998]

Allocating a reference-counted buffer with input.copy() introduces a leak, as writeBytes() does not release it. This temporary copy is leaked even on a successful test.

Consider using encrypted.writeBytes(input, input.readerIndex(), input.readableBytes()) instead. This avoids the extra allocation and preserves the reader index. Additionally, enclosing the retained-buffer cleanup in a finally block would prevent leaks from failed assertions.

[SongOf]

Fixed in 1ba0e84. The mock now uses encrypted.writeBytes(input, input.readerIndex(), input.readableBytes()), so there is no intermediate copy and the reader index is preserved. The retained-buffer cleanup in the oversized-batch tests is in a finally via ReferenceCountUtil.safeRelease, so a failed assertion no longer leaks the buffer.