support limit message size produced by protocol handlers by maxMessag…

[gaozhangmin]

Motivation

See #13170 .
Also, Protocols handlers like KOP should make changes to add and decrease pendingBytes from BrokerService.

Modifications

Add a LongAdder totalPendingBytes in BrokerService to record total pending message size.

Documentation

Check the box below and label this PR (if you have committer privilege).

Need to update docs?

• doc-required

  (If you need help on updating docs, create a doc issue)

• no-need-doc

  (Please explain why)

• doc

  (If this PR contains doc changes)

[github-actions]

@gaozhangmin:Thanks for your contribution. For this PR, do we need to update docs?
(The PR template contains info about doc, which helps others know more about the changes. Can you provide doc-related info in this and future PR descriptions? Thanks)

[github-actions]

@gaozhangmin:Thanks for providing doc info!

[wangjialing218]

There could have a case when we have both pulsar and kafka producer send msgs:

1. totalPendingBytes exceed maxMessagePublishBufferSizeInMB and pulsar broker and KoP both stop read from producer.
2. Pulsar have completed all it's pending msgs, but totalPendingBytes still exceed resumeThresholdPendingBytes since KoP may hold lots of pending buffer, pulsar will not start to read from pulsar producer.
3. KoP completed some pending msgs and totalPendingBytes become lower than resumeThresholdPendingBytes, KoP will start to read from kafka producer. But pulsar broker may still not start to read from pulsar producer because completedSendOperation will never called since all pulsar msgs are sent completed.

Could you please consider how to avoid this happen

[gaozhangmin]

Why the previous design had no this problem? I'm little confused. @wangjialing218

[gaozhangmin]

I got the reason, I will try fix this case.

[wangjialing218]

As previous design，totalPendingBytes is only used by broker to record the msg bytes received from pulsar producer, so broker could resume reading from producer when totalPendingBytes become lower than resumeThresholdPendingBytes.
This PR will make totalPendingBytes shared with KoP, so this problem could happen.

[wangjialing218]

We do not have to do LongAdder.sum() each time when add or decrease pending bytes, since it's a little heavy CPU work.
We could do sum() in another scheduler task such as 50ms each time, and then inform ServerCnx in broker and protocol handlers to stop or resume reading by listener.

[gaozhangmin]

@codelipenghui @eolivelli @wangjialing218 @Jason918 @BewareMyPower PTAL

[Jason918]

It's better that we can describe more details about the downside of this solution.

[Jason918]

I would prefer a better data structure than CopyOnWriteArrayList if we have a lot of producers.

[gaozhangmin]

/pulsarbot run-failure-checks

[gaozhangmin]

/pulsarbot run-failure-checks

[hangc0276]

@gaozhangmin Thanks for your contribution.
Before Pulsar 2.8.0, the broker publish throttle policy is similar to your implementation, it has a shortcoming of suddenly exceed the limit and will lead to broker direct memory OOM. After Pulsar 2.8.0, we introduced the throttle policy by io thread, it can limit the throughput of pulsar protocol, but out of control of other protocols, like KOP.

IMO, it's better to throttle in KOP side, do you have any other ideas? @BewareMyPower

[BewareMyPower]

@hangc0276 For now I didn't have much context. AFAIK, this PR is an implementation of idea from #12959. Could you answer hang's question? @wangjialing218

[wangjialing218]

@hangc0276 You may mentioned #7406, which introduced the throttle policy by io thread.
This PR make more effective use of memory, and could also improve the points mentioned in #7406, except this:
-- There is a delay for detecting the memory over-commit, due to the background task running periodically
As idle connection will not call LoggAdder.add, we only need to count up the bytes from active connection in LoggAdder.sum. We could do the detecting more frequently to avoid OOM, such as 10ms per time.

IMO, it's better to throttle in KOP side, do you have any other ideas?

For this point, we could make sure broker and KoP share the IO thread pool, and do same throttle policy by io thread in KoP side. @BewareMyPower
I notice there is a configuration useSeparateThreadPoolForProtocolHandlers with default value true. I wonder if there is any disadvantage when set this configuration false.

[BewareMyPower]

I notice there is a configuration useSeparateThreadPoolForProtocolHandlers with default value true. I wonder if there is any disadvantage when set this configuration false.

The false configuration is the previous behavior that the potential deadlocks at KoP side might also block broker's IO thread.