[Fix][broker] Limit replication rate based on bytes

[nodece]

Motivation

When the replicator rate limit value based on bytes is less than the org.apache.pulsar.broker.service.persistent.PersistentReplicator#readMaxSizeBytes, this will cause the outgoing bytes to exceed the limit value.

The goal is to make outgoing bytes as close to the rate limit value as possible.

Modifications

• org.apache.pulsar.broker.service.persistent.PersistentReplicator#getAvailablePermits returns the available messages and bytes, and then reads the entries based on that.

Verifying this change

The test has been added.

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

[lhotari]

It's better to use a record than to use Pair. Pair results in hard-to-read code. An additional reason here is that Integer and Long are boxed objects which are adding additional overhead compared to a record.

[nodece]

I can add a class to hold the available messages and bytes, so like:

    @Data
    @AllArgsConstructor
    private static class AvailablePermits {
        private int messages;
        private long bytes;
    }

record requires the JDK 14, we cannot cherry-pick this PR to the previous version.

[nodece]

Ping @lhotari

[nodece]

/pulsarbot rerun-failure-checks

[codecov-commenter]

Codecov Report

Attention: Patch coverage is 65.38462% with 9 lines in your changes are missing coverage. Please review.

Project coverage is 73.18%. Comparing base (bbc6224) to head (a3277c3).
Report is 264 commits behind head on master.

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #22674      +/-   ##
============================================
- Coverage     73.57%   73.18%   -0.40%     
- Complexity    32624    32849     +225     
============================================
  Files          1877     1889      +12     
  Lines        139502   141370    +1868     
  Branches      15299    15515     +216     
============================================
+ Hits         102638   103457     +819     
- Misses        28908    29911    +1003     
- Partials       7956     8002      +46     

Flag	Coverage Δ
inttests	27.52% <42.30%> (+2.93%)	⬆️
systests	24.70% <0.00%> (+0.38%)	⬆️
unittests	72.20% <65.38%> (-0.65%)	⬇️

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

Files	Coverage Δ
...roker/service/persistent/PersistentReplicator.java	67.54% <65.38%> (-1.34%)	⬇️

... and 353 files with indirect coverage changes