[New Feature] Add new configuration options which allows broker to use a bounded Jersey ThreadPool

[MeihanLi]

[New Feature] Add new configuration options which allows broker to use a bounded Jersey ThreadPool

Currently Broker uses Jersey default unbounded thread pool to process async requests and it uses No-Op RejectedExecutionHandler.

When a broker is serving very high QPS, a significant amount of threads will be accumulated on the broker. We have seen ~25k threads accumulated in a short time because of this and brokers were taken down very quickly. This is also the root cause of #9019.

This PR
(1) Implements a custom provider BrokerManagedAsyncExecutorProvider by extending ThreadPoolExecutorProvider, which will be automatically picked by Jersey. When async requests can not be accommodated, they will get rejected and customers will receive ERROR code 503 (Service Unavailable).

(2) Add new configuration options below which allow us to use the bounded thread pool and allocate capacities for it. By default, it is disabled.
pinot.broker.enable.bounded.http.async.executor
pinot.broker.http.async.executor.max.pool.size
pinot.broker.http.async.executor.core.pool.size
pinot.broker.http.async.executor.queue.size

By enabling the new configs, endpoints POST /query/sql and GET /query/sql will be impacted.

Test

Tested locally, one example response for a rejected request is

HTTP/1.1 503 Service Unavailable
Content-Type: application/json
Connection: close
Content-Length: 220

Pinot Broker thread pool can not accommodate more requests now. Request is rejected from java.util.concurrent.ThreadPoolExecutor@79c8b52f[Running, pool size = 1, active threads = 1, queued tasks = 1, completed tasks = 6]

The error log emitted by BrokerManagedAsyncExecutorProvider is:
2023/04/14 16:47:26.146 ERROR [BrokerManagedAsyncExecutorProvider] [grizzly-http-server-15] Task java.util.concurrent.FutureTask@dafa6aa[Not completed, task = java.util.concurrent.Executors$RunnableAdapter@74af7b05[Wrapped task = org.glassfish.jersey.server.ServerRuntime$AsyncResponder$2@58e4de25]] rejected from java.util.concurrent.ThreadPoolExecutor@1bebdf0b[Running, pool size = 1, active threads = 1, queued tasks = 1, completed tasks = 6]

[codecov-commenter]

Codecov Report

Merging #10614 (827c537) into master (13a792f) will increase coverage by 1.36%.
The diff coverage is 70.58%.

@@             Coverage Diff              @@
##             master   #10614      +/-   ##
============================================
+ Coverage     69.03%   70.40%   +1.36%     
+ Complexity     6499     5629     -870     
============================================
  Files          2106     2106              
  Lines        113010   113674     +664     
  Branches      17027    17216     +189     
============================================
+ Hits          78015    80029    +2014     
+ Misses        29515    28050    -1465     
- Partials       5480     5595     +115     

Flag	Coverage Δ
integration1	24.74% <8.82%> (?)
integration2	24.08% <8.82%> (+0.03%)	⬆️
unittests1	67.88% <100.00%> (-0.15%)	⬇️
unittests2	14.03% <58.82%> (+0.12%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files	Coverage Δ
...pinot/broker/broker/BrokerAdminApiApplication.java	78.48% <20.00%> (-8.48%)	⬇️
...ker/broker/BrokerManagedAsyncExecutorProvider.java	90.00% <90.00%> (ø)
...a/org/apache/pinot/common/metrics/BrokerMeter.java	100.00% <100.00%> (ø)
...va/org/apache/pinot/spi/utils/CommonConstants.java	27.05% <100.00%> (+2.66%)	⬆️

... and 183 files with indirect coverage changes

📣 We’re building smart automated test selection to slash your CI/CD build times. Learn more

[ankitsultana]

Thanks Christina for addressing all the feedback!

[wirybeaver]

queue size actually is bounded by memory. but I think availableProcessors() * 2 is OK.

[jasperjiaguo]

Would the default queue size of availableProcessors() * 2 be too small? During spiky request we may drop requests prematurely?

[MeihanLi]

Hi Jasper, I didn't test the default values so they might be small. Can you share what might be a better number here?

[MeihanLi]

@jasperjiaguo I set the default value to Integer.MAX_VALUE, let me know if this looks good to you. The change is in the last commit.

[jasperjiaguo]

Yes that looks good to me. Thanks!

[Jackie-Jiang]

LGTM otherwise

[jasperjiaguo]

LGTM otherwize

[jasperjiaguo]

Would the default queue size of availableProcessors() * 2 be too small? During spiky request we may drop requests prematurely?

[MeihanLi]

LGTM otherwise

Thanks @Jackie-Jiang , comments are all addressed. Can you help take another look. Thanks

[Jackie-Jiang]

@MeihanLi Can you help update the pinot documentation to include this new feature?

[MeihanLi]

@MeihanLi Can you help update the pinot documentation to include this new feature?

Thanks Jackie for merging it, documentation is updated here: pinot-contrib/pinot-docs#164

[Jackie-Jiang]

@MeihanLi By reading the pinot doc, I just realized that we already have a set of configs (pinot.broker.http.server.thread.pool.corePoolSize and pinot.broker.http.server.thread.pool.maxPoolSize) for http servers, which can be applied to all components. Are they for the same purpose? If so, are we able to integrate the new config at the same place? Currently they are applied in 2 different places, which is quite hard to manage.

cc @apucher who has more context on how to config the rest server

[ankitsultana]

@Jackie-Jiang : Yeah I had added those. Those configs control the thread-pool for the sync APIs afaiu. We mainly use them for tuning the thread-pool size for the pinot-controllers.

I think we don't really need to support that config for brokers. We could consider removing support for it to reduce confusion.