[PIP-111] Add messages produced by Protocol Handler When checking maxMessagePublishBufferSizeInMB #13170

Motivation

MaxMessagePublishBufferSizeInMB is used to limit the direct memory usage for reading messages from producers, default value is 1/2 of direct memory.
The problem is:
If we enable any Protocol Handler in broker such as KoP, the direct memory to buffer the pending messages from Kafka producers is not limited by maxMessagePublishBufferSizeInMB. And if we have Pulsar producers and Kafka producers sending message together with higher throughput than BookKeeper IO, broker OOM will happen.

Goal

When handling producer message, we could check if pending bytes from pulsar and protocol handler exceed maxMessagePublishBufferSizeInMB, if yes, stop reading from producer both in broker and Protocol Handler side.

API Changes

using a LongAdder in BrokerService to record the current pending bytes for entire broker

Implementation

To make effective use of memory, using a LongAdder in BrokerService to record the current pending bytes for entire broker, include both pulsar pending messages and Protocol Handler's pending messages.

Reject Alternatives

No alternatives yet.