[fix][broker] Release the compaction buffers and permits on flush failures

[nodece]

Motivation

The compaction flush path leaks buffers and permits on failure:

• StrategicTwoPhaseCompactor.flushBatchMessage(): the serialized batch buffer is owned solely by the local variable once toByteBuf() returned (the container has already cleared itself), but a failure before the BookKeeper hand-off completes — an interrupted outstanding.acquire(), a throwing stats update, or asyncAddEntry failing synchronously — only discarded the (already empty) container: the serialized buffer leaked on every failed flush.
• The outstanding-writes permit taken before such a failure was never returned either (the write callback that owns the release never runs), permanently shrinking the limit until compaction stalls in acquire().
• RawBatchMessageContainerImpl.toByteBuf(): the output buffer is allocated before the header writes, but a failure between the allocation and the return left it orphaned — the finally released the intermediate buffers only.

Modifications

• flushBatchMessage() releases the serialized buffer when the hand-off never happened (it is nulled after the successful hand-off, when BookKeeper owns it) and returns the permit when it was acquired — an acquire() that itself throws must not release one it never took, or the outstanding-writes limit silently grows.
• The write callback's metrics update is caught: the callback can run inline inside asyncAddEntry, and an exception escaping it would reach the caller's catch, releasing the permit and the buffer a second time after the write already took them. With the callback unable to throw, an exception from asyncAddEntry itself can only mean the callback never ran — verified against BookKeeper 4.18.0, whose synchronous throws in doAsyncAddEntry all precede any callback execution — which is exactly the case the caller's catch handles.
• toByteBuf() releases the output buffer in its finally as well, nulled on the success path where its ownership moved to the returned buffer; its allocation goes through the instance allocator (the default constructor already uses PulsarByteBufAllocator.DEFAULT), which is what lets the regression test track it.

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows: RawBatchMessageContainerImplTest#testToByteBufReleasesTheOutputBufferWhenSerializationFails (fails with "expected [0] but found [1]" on the previous code). The compactor fixes have no direct test seam (private flush method, inline container); the reasoning is stated above.

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 on the compaction flush path, no user-facing behavior or configuration change)

[lhotari]

• the output buffer is allocated before the header writes, but a failure between the allocation and the return left it orphaned

a meta-question: are you running into allocation failures? Perhaps that should be addressed as the primary solution instead of handling all possible code paths where allocation could fail. A recommended setting is to crash the broker when memory allocation fails instead of trying to continue operations.

We do have this setting in bin/pulsar by default, but perhaps it doesn't work?

pulsar/bin/pulsar

Lines 299 to 302 in 7a3ec43

	# These two settings work together to ensure the Pulsar process exits immediately and predictably
	# if it runs out of either Java heap memory or its internal off-heap memory,
	# as these are unrecoverable errors that require a process restart to clear the faulty state and restore operation
	OPTS="-XX:+ExitOnOutOfMemoryError -Dpulsar.allocator.exit_on_oom=true $OPTS"

There's a lot of discussion on #25021 about addressing Netty allocator problems.
In particular, the comment #25021 (comment)
more in #25522 (comment)

One of the gaps in the broker is this one:
#24926
The problem itself is also referred in the OOM issue.

Focusing on fixing the root cause could be more effective in the long term.

[lhotari]

One source of OOM problems in dispatching might be caused by the default dispatcherDispatchMessagesInSubscriptionThread=true setting. The default causes slower performance and higher resource consumption as well. There has been discussions with @merlimat to change the default to dispatcherDispatchMessagesInSubscriptionThread=false.

The way how dispatcherDispatchMessagesInSubscriptionThread=true causes problems for memory consumption is that the entries are held until the entries are dispatched (and written to the socket). This is also related to #24926. If the entries are sitting in the outgoing buffer, the memory won't be released.

[lhotari]

PR to change dispatcherDispatchMessagesInSubscriptionThread=false as the default: #26578