Group and flush add-responses after journal sync

[merlimat]

Motivation

Note: this is stacked on top of #3830 & #3835

This change improves the way the AddRequests responses are send to client.

The current flow is:

• The journal-force-thread issues the fsync on the journal file
• We iterate over all the entries that were just synced and for each of them:
  1. Trigger channel.writeAndFlus()
  2. This will jump on the connection IO thread (Netty will use a write() to eventfd to post the task and wake the epoll)
  3. Write the object in the connection and trigger the serialization logic
  4. Grab a ByteBuf from the pool and write ~20 bytes with the response
  5. Write and flush the buffer on the channel
  6. With the flush consolidator we try to group multiple buffer into a single writev() syscall, though each call will have a long list of buffer, making the memcpy inefficient.
  7. Release all the buffers and return them to the pool

All these steps are quite expensive when the bookie is receiving a lot of small requests.

This PR changes the flow into:

1. journal fsync
2. go through each request and prepare the response into a per-connection ByteBuf which is not written on the channel as of yet
3. after preparing all the responses, we flush them at once: Trigger an event on all the connections that will write the accumulated buffers.

The advantages are:

1. 1 ByteBuf allocated per connection instead of 1 per request
   1. Less allocations and stress of buffer pool
   2. More efficient socket write() operations
2. 1 task per connection posted on the Netty IO threads, instead of 1 per request.

[hangc0276]

Good job! Please rebase the master.

[codecov-commenter]

Codecov Report

Merging #3837 (138ffb6) into master (73c5a0e) will increase coverage by 13.69%.
The diff coverage is 96.29%.

@@              Coverage Diff              @@
##             master    #3837       +/-   ##
=============================================
+ Coverage     54.67%   68.36%   +13.69%     
- Complexity     5346     6774     +1428     
=============================================
  Files           473      473               
  Lines         40910    40960       +50     
  Branches       5232     5240        +8     
=============================================
+ Hits          22366    28003     +5637     
+ Misses        16462    10706     -5756     
- Partials       2082     2251      +169     

Flag	Coverage Δ
bookie	39.70% <83.33%> (-0.10%)	⬇️
client	44.14% <83.33%> (?)
remaining	29.70% <44.44%> (?)
replication	41.34% <92.59%> (+0.05%)	⬆️
tls	20.95% <88.88%> (?)

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

Impacted Files	Coverage Δ
...main/java/org/apache/bookkeeper/bookie/Bookie.java	75.00% <ø> (ø)
...pache/bookkeeper/proto/BookieRequestProcessor.java	74.60% <85.71%> (+24.34%)	⬆️
.../apache/bookkeeper/proto/BookieRequestHandler.java	86.36% <95.83%> (+10.36%)	⬆️
.../java/org/apache/bookkeeper/bookie/BookieImpl.java	71.45% <100.00%> (+1.45%)	⬆️
...ain/java/org/apache/bookkeeper/bookie/Journal.java	79.67% <100.00%> (-0.33%)	⬇️
...g/apache/bookkeeper/proto/BookieProtoEncoding.java	80.93% <100.00%> (+6.23%)	⬆️
...java/org/apache/bookkeeper/proto/BookieServer.java	71.28% <100.00%> (+2.28%)	⬆️
...g/apache/bookkeeper/proto/WriteEntryProcessor.java	76.05% <100.00%> (+6.61%)	⬆️
...a/org/apache/bookkeeper/proto/ResponseBuilder.java	72.72% <0.00%> (-18.19%)	⬇️
.../java/org/apache/bookkeeper/bookie/SyncThread.java	69.69% <0.00%> (-3.04%)	⬇️
... and 169 more

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

[zymap]

Do we need to record journalStats.getJournalFlushStats() before trigger flushPendingResponses? That would lead the JournalFlushStats to include response time.

[merlimat]

The flush would be very fast though. It's only posting an event on the channel group, not writing the actuals responses on the channels.

[horizonzy]

Here we trigger all channels to flush pendingSendResponses, shall we only pick the requestHandler which is written to the logFile to flush pendingSendResponses, it may reduce the unnecessary iteration in the loop.

    public void flushPendingResponses() {
        for (Channel c : allChannels) {
            c.pipeline().fireUserEventTriggered(BookieRequestHandler.EVENT_FLUSH_ALL_PENDING_RESPONSES);
        }
    }

If there are 1000 channels, maybe only 2 channels need to flush pendingSendResponses