[Broker] Fix broker dispatch byte rate limiter. #11135

congbobo184 · 2021-06-28T15:33:09Z

Motivation

fix #11044
now dispatcher byte rate limit don't limit every cursor read. When cursor read always use ServiceConfiguration.dispatcherMaxReadSizeBytes to read. It will cause that dispatcher read entries by ServiceConfiguration.dispatcherMaxReadSizeBytes to read every time.

implement

when cursor read entries size need to calculate, the calculate result by dispatcher bytes limiter.

Verifying this change

Add the tests for it

Does this pull request potentially affect one of the following parts:
If yes was chosen, please highlight the changes

Dependencies (does it add or upgrade a dependency): (no)
The public API: (no)
The schema: (no)
The default values of configurations: (no)
The wire protocol: (no)
The rest endpoints: (no)
The admin cli options: (no)
Anything that affects deployment: (no)

codelipenghui · 2021-06-29T01:16:43Z

.../java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java

@@ -276,8 +278,10 @@ public synchronized void readMoreEntries() {
        }
    }

-    protected int calculateNumOfMessageToRead(int currentTotalAvailablePermits) {
+    // left pair is messagesToRead, right pair is bytesToRead
+    protected MutablePair<Integer, Integer> calculateToRead(int currentTotalAvailablePermits) {


Why need MutablePair here? we'd better do not allow to modify it outside?

codelipenghui · 2021-06-29T02:08:16Z

...va/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java

@@ -349,7 +352,7 @@ protected void readMoreEntries(Consumer consumer) {
        }
    }

-    protected int calculateNumOfMessageToRead(Consumer consumer) {
+    protected MutablePair<Integer, Integer> calculateToRead(Consumer consumer) {


Is it possible to move to AbstractBaseDispatcher? Looks many duplicated code with the PersistentDispatcherMultipleConsumers.

codelipenghui · 2021-06-29T02:13:04Z

pulsar-broker/src/test/java/org/apache/pulsar/client/api/MessageDispatchThrottlingTest.java

        for (int i = 0; i < numProducedMessages; i++) {
-            producer.send(new byte[byteRate / 10]);
+            producer.send(new byte[99]);


It's better to use asynchronous API? If we use synchronous API, this means for each message we need to ensure the publish latency is lower than 50ms, otherwise it should be a flaky test.

And what's the reason change to new byte[99]

codelipenghui · 2021-06-29T02:13:48Z

pulsar-broker/src/test/java/org/apache/pulsar/client/api/MessageDispatchThrottlingTest.java

        }

-        latch.await();
-        Assert.assertEquals(totalReceived.get(), numProducedMessages);
+        Awaitility.await().atLeast(3, TimeUnit.SECONDS)


Why we need atLeast here?

we need to check when the time less than 3 second consumer con't receive messages more than 6.

eolivelli · 2021-06-29T11:02:10Z

@lhotari PTAL

315157973

Why not use ImmutablePair ?

anvinjain · 2021-06-29T09:31:46Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java

@@ -234,6 +238,19 @@ public void resetCloseFuture() {
        // noop
    }

+    protected static Pair<Integer, Integer> calculateToRead(int messagesToRead, int availablePermitsOnMsg,


Maybe a better name like computeReadLimits ?
Can parameters be ordered as: messagesToRead, availableMsgPermits, bytesToRead, availableBytePermits just to have a consistent grouping

good suggestion.

anvinjain · 2021-06-29T09:45:45Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java

+        }
+
+        if (availablePermitsOnByte > 0) {
+            bytesToRead = availablePermitsOnByte;


dispatcherMaxReadSizeBytes was used to limit max number of bytes that can be read from bookies in one cursor read op. With this change if user configures a dispatch rate of say 100MB/s, it will mean that bytesToRead can be 100MB (since we do not do Math.min here)

good catch!

anvinjain · 2021-06-29T09:53:05Z

.../java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java

-                    }
+                    Pair<Integer, Integer> calculateResult = calculateToRead(messagesToRead,
+                            (int) topicRateLimiter.getAvailableDispatchRateLimitOnMsg(),
+                            (int) topicRateLimiter.getAvailableDispatchRateLimitOnByte(), bytesToRead);


Should we do type conversion for bytes from long to int? Could cause bugs when user sets dispatch rate higher than 2GB (unlikely but possible)

anvinjain · 2021-06-29T17:20:46Z

.../java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java

@@ -263,8 +264,7 @@ public synchronized void readMoreEntries() {
                            consumerList.size());
                }
                havePendingRead = true;
-                cursor.asyncReadEntriesOrWait(messagesToRead, serviceConfig.getDispatcherMaxReadSizeBytes(),
-                        this,
+                cursor.asyncReadEntriesOrWait(messagesToRead, bytesToRead, this,


bytesToRead can potentially be -1 now (like at line 315 when new MutablePair<>(-1, -1); is returned) when dispatch limit is not set. This changes the default behaviour because earlier it was capped by serviceConfig.getDispatcherMaxReadSizeBytes()

in line 230, when bytesToRead or messageToRead is -1, it will not process read operation.

congbobo184

thank for you review, I left some question and change the code. please review again. thanks.

congbobo184 · 2021-06-30T01:26:05Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java

@@ -234,6 +238,19 @@ public void resetCloseFuture() {
        // noop
    }

+    protected static Pair<Integer, Integer> calculateToRead(int messagesToRead, int availablePermitsOnMsg,


good suggestion.

congbobo184 · 2021-06-30T01:32:25Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java

+        }
+
+        if (availablePermitsOnByte > 0) {
+            bytesToRead = availablePermitsOnByte;


good catch!

congbobo184 · 2021-06-30T01:36:25Z

.../java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java

@@ -263,8 +264,7 @@ public synchronized void readMoreEntries() {
                            consumerList.size());
                }
                havePendingRead = true;
-                cursor.asyncReadEntriesOrWait(messagesToRead, serviceConfig.getDispatcherMaxReadSizeBytes(),
-                        this,
+                cursor.asyncReadEntriesOrWait(messagesToRead, bytesToRead, this,


in line 230, when bytesToRead or messageToRead is -1, it will not process read operation.

congbobo184 · 2021-06-30T02:34:38Z

Why not use ImmutablePair ?

changed

anvinjain · 2021-06-30T06:01:14Z

.../java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java

+                    Pair<Integer, Long> calculateResult = computeReadLimits(messagesToRead,
+                            (int) topicRateLimiter.getAvailableDispatchRateLimitOnMsg(),
+                            topicRateLimiter.getAvailableDispatchRateLimitOnByte(), bytesToRead);


You have not changed the order of arguments here (after changing the order in the function definition). It should be

computeReadLimits( messagesToRead, (int) topicRateLimiter.getAvailableDispatchRateLimitOnMsg(), bytesToRead, topicRateLimiter.getAvailableDispatchRateLimitOnByte());

Please fix at other places too where this function is called.

done, could you please review again? if don't have any other comment, please approve this pr then we will merge it :)

codelipenghui · 2021-07-01T00:59:22Z

@anvinjain Do you want to take a final look?

## Motivation fix #11044 now dispatcher byte rate limit don't limit every cursor read. When cursor read always use `ServiceConfiguration.dispatcherMaxReadSizeBytes` to read. It will cause that dispatcher read entries by `ServiceConfiguration.dispatcherMaxReadSizeBytes` to read every time. (cherry picked from commit ce6be12)

The PR #11135 couldn't be cherry-picked to branch-2.7, because there are too many conflicts. ## Motivation fix #11044 now dispatcher byte rate limit don't limit every cursor read. When cursor read always use `ServiceConfiguration.dispatcherMaxReadSizeBytes` to read. It will cause that dispatcher read entries by `ServiceConfiguration.dispatcherMaxReadSizeBytes` to read every time. ## implement when cursor read entries size need to calculate, the calculate result by dispatcher bytes limiter.

## Motivation fix apache#11044 now dispatcher byte rate limit don't limit every cursor read. When cursor read always use `ServiceConfiguration.dispatcherMaxReadSizeBytes` to read. It will cause that dispatcher read entries by `ServiceConfiguration.dispatcherMaxReadSizeBytes` to read every time.

[Broker] Fix broker dispatch byte rate limiter.

78654bf

congbobo184 requested review from codelipenghui, eolivelli, gaoran10, merlimat, rdhabalia, 315157973, hangc0276 and sijie June 28, 2021 15:33

congbobo184 self-assigned this Jun 28, 2021

congbobo184 added the release/2.8.1 label Jun 28, 2021

codelipenghui reviewed Jun 29, 2021

View reviewed changes

codelipenghui added this to the 2.9.0 milestone Jun 29, 2021

codelipenghui added the release/2.7.3 label Jun 29, 2021

congbo added 2 commits June 29, 2021 15:48

Fix some comment.

f2d0d8c

Fix codestyle

5deea18

codelipenghui approved these changes Jun 29, 2021

View reviewed changes

315157973 approved these changes Jun 29, 2021

View reviewed changes

anvinjain reviewed Jun 29, 2021

View reviewed changes

fix some comment

7588829

congbobo184 commented Jun 30, 2021

View reviewed changes

anvinjain suggested changes Jun 30, 2021

View reviewed changes

fix some comment

820380f

sijie approved these changes Jun 30, 2021

View reviewed changes

anvinjain approved these changes Jul 5, 2021

View reviewed changes

codelipenghui merged commit ce6be12 into apache:master Jul 6, 2021

codelipenghui added the cherry-picked/branch-2.8 Archived: 2.8 is end of life label Jul 7, 2021

gaoran10 mentioned this pull request Jul 7, 2021

[Branch-2.7] Fix broker dispatch byte rate limiter #11249

Merged

codelipenghui added the cherry-picked/branch-2.7 Archived: 2.7 is end of life label Jul 12, 2021

congbobo184 deleted the congbobo184_fix_dispatcher_byte_limit branch March 24, 2022 04:59

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Broker] Fix broker dispatch byte rate limiter. #11135

[Broker] Fix broker dispatch byte rate limiter. #11135

congbobo184 commented Jun 28, 2021

codelipenghui Jun 29, 2021

codelipenghui Jun 29, 2021

codelipenghui Jun 29, 2021

codelipenghui Jun 29, 2021

codelipenghui Jun 29, 2021

congbobo184 Jun 29, 2021

eolivelli commented Jun 29, 2021

315157973 left a comment

anvinjain Jun 29, 2021

congbobo184 Jun 30, 2021

anvinjain Jun 29, 2021

congbobo184 Jun 30, 2021

anvinjain Jun 29, 2021

anvinjain Jun 29, 2021

congbobo184 Jun 30, 2021

congbobo184 left a comment

congbobo184 Jun 30, 2021

congbobo184 Jun 30, 2021

congbobo184 Jun 30, 2021

congbobo184 commented Jun 30, 2021

anvinjain Jun 30, 2021

congbobo184 Jun 30, 2021 •

edited

Loading

codelipenghui commented Jul 1, 2021

[Broker] Fix broker dispatch byte rate limiter. #11135

[Broker] Fix broker dispatch byte rate limiter. #11135

Conversation

congbobo184 commented Jun 28, 2021

Motivation

implement

Verifying this change

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

eolivelli commented Jun 29, 2021

315157973 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

congbobo184 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

congbobo184 commented Jun 30, 2021

Choose a reason for hiding this comment

congbobo184 Jun 30, 2021 • edited Loading

Choose a reason for hiding this comment

codelipenghui commented Jul 1, 2021

congbobo184 Jun 30, 2021 •

edited

Loading