[broker] Optimize blocking backlogQuotaCheck to non-blocking in ServerCnx#handleProducer #12874

Jason918 · 2021-11-18T13:39:23Z

Motivation

Currently, when broker receive "Producer" command, it will check the topic if "isBacklogQuotaExceeded".
While in PersistentTopic#isBacklogQuotaExceeded, isTimeBacklogExceeded is used, in which will turns to a blocking operation if "isPreciseTimeBasedBacklogQuotaCheck" is set as true.

The blocking operations in pulsar io threads may impact broker performance, this PR optimized this blocking procedure to async mode.

Modifications

Add "CompletableFuture checkTimeBacklogExceeded()" in PersistentTopic for the async check procedure.
Update corresponding method calls to async mode in ServerCnx#handleProducer.

Verifying this change

Make sure that the change passes the CI checks.

This change is already covered by existing tests, such as org.apache.pulsar.broker.admin.TopicPoliciesTest and org.apache.pulsar.broker.service.BacklogQuotaManagerTest

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)

Documentation

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

Need to update docs?

no-need-doc

Only code optimize.

Jason918 · 2021-11-19T02:56:57Z

/pulsarbot run-failure-checks

Jason918 · 2021-11-19T08:42:08Z

/pulsarbot run-failure-checks

Jason918 · 2021-11-19T09:56:08Z

/pulsarbot run-failure-checks

Jason918 · 2021-11-19T13:08:54Z

/pulsarbot run-failure-checks

Jason918 · 2021-11-19T14:53:00Z

/pulsarbot run-failure-checks

Jason918 · 2021-11-20T02:12:36Z

@codelipenghui @hangc0276 @congbobo184 Please help take a look.

codelipenghui · 2021-11-22T13:52:44Z

@Jason918 Do you want the resolve the conflicts?

Jason918 · 2021-11-22T15:02:51Z

@Jason918 Do you want the resolve the conflicts?

Resolved, sorry for missing this.

Jason918 · 2021-11-23T11:11:20Z

/pulsarbot run-failure-checks

michaeljmarshall

This is a good change. I think we should probably discuss it on the mailing list before moving forward since it includes breaking changes to the Topic interface. @eolivelli, @codelipenghui, @hangc0276 - do you agree, or is it minor enough that we can move forward here?

eolivelli · 2021-11-23T21:43:44Z

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

+                        topic.checkBacklogQuotaExceeded(producerName, BacklogQuotaType.message_age));
+                backlogQuotaCheckFuture.exceptionally(throwable -> {
+                    //throwable should be CompletionException holding TopicBacklogQuotaExceededException
+                    BrokerServiceException.TopicBacklogQuotaExceededException exception =


What happens if there is something else?
A ClassCastException or a NPE.
This will break the system

I believe we should handle the case in which this expectation is not met

Two situations:

If preciseTimeBasedBacklogQuotaCheck set as true (default is false) and some other constrains met, this will be execute in the callback thread of ManagedLedgerImpl#asyncReadEntry(...).

Otherwise everything is executed in previous thread. Nothing async.

eolivelli · 2021-11-23T21:46:30Z

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

+                CompletableFuture<Void> backlogQuotaCheckFuture = CompletableFuture.allOf(
+                        topic.checkBacklogQuotaExceeded(producerName, BacklogQuotaType.destination_storage),
+                        topic.checkBacklogQuotaExceeded(producerName, BacklogQuotaType.message_age));
+                backlogQuotaCheckFuture.exceptionally(throwable -> {


Which thread will execute this code?

I am changing the outer thenAccept to thenCompose, the outside exceptionally will handle all unexpected exceptions.

eolivelli · 2021-11-23T21:47:51Z

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


-                disableTcpNoDelayIfNeeded(topicName.toString(), producerName);
+                backlogQuotaCheckFuture.thenRun(() -> {


Which thread will execute this code?

The current thread (in case we reach here and the futures are already completed) or some other thread on completion of the one of the two futures above.

I am not sure we have control over what's happening here

The trigger thread is the same as the above backlogQuotaCheckFuture.exceptionally, it's current thread or ReadEntry call back thread.

eolivelli · 2021-11-23T21:49:28Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java

@@ -2518,17 +2528,26 @@ public boolean isSizeBacklogExceeded() {
     * @return determine if backlog quota enforcement needs to be done for topic based on time limit
     */
    public boolean isTimeBacklogExceeded() {
+        try {
+            return checkTimeBacklogExceeded().get();
+        } catch (Throwable e) {


This is a code smell

This is just for keeping the same behavior of isTimeBacklogExceeded as before. This method is called somewhere else.

The main problem with this approach is that it completely hides from the caller the fact that the method is internally blocking. We have many other examples of this in the code base

I get it. Thx.
I will remove this isTimeBacklogExceeded.

Jason918 · 2021-11-24T06:06:17Z

/pulsarbot run-failure-checks

Jason918 · 2021-11-25T03:25:17Z

/pulsarbot run-failure-checks

Jason918 · 2021-12-06T06:21:38Z

@eolivelli PTAL

codelipenghui · 2022-01-18T06:35:38Z

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

+                CompletableFuture<Void> backlogQuotaCheckFuture = CompletableFuture.allOf(
+                        topic.checkBacklogQuotaExceeded(producerName, BacklogQuotaType.destination_storage),
+                        topic.checkBacklogQuotaExceeded(producerName, BacklogQuotaType.message_age));
+                backlogQuotaCheckFuture.exceptionally(throwable -> {


thenCompose return the backlogQuotaCheckFuture to the next stage, is it better to handle the TopicBacklogQuotaExceededException here https://github.com/apache/pulsar/pull/12874/files#diff-1e0e8195fb5ec5a6d79acbc7d859c025a9b711f94e6ab37c94439e99b3202e84R1262 ?

@codelipenghui Moved TopicBacklogQuotaExceededException handling outside. PTAL.

codelipenghui · 2022-01-18T06:36:06Z

@eolivelli Please help review this PR again

Jason918 · 2022-01-18T12:05:54Z

/pulsarbot run-failure-checks

codelipenghui · 2022-01-21T04:18:21Z

@eolivelli Please help review this PR again

eolivelli

LGTM

### Motivation We should only send the error response to the client when the code is able to complete the `producerFuture`. This logic is described here: https://github.com/apache/pulsar/blob/2285d02aa9957af7877b9d3d3c628a750d813ca7/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java#L1286-L1293 Edit: in a previous version of this motivation section, I attributed the current behavior to #12874. That PR did not introduce this behavior, though. ### Modifications * Move the response to the client into a conditional block that only runs when this section of the code is able to complete the future.

…he#13949) ### Motivation We should only send the error response to the client when the code is able to complete the `producerFuture`. This logic is described here: https://github.com/apache/pulsar/blob/2285d02aa9957af7877b9d3d3c628a750d813ca7/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java#L1286-L1293 Edit: in a previous version of this motivation section, I attributed the current behavior to apache#12874. That PR did not introduce this behavior, though. ### Modifications * Move the response to the client into a conditional block that only runs when this section of the code is able to complete the future.

github-actions bot assigned Jason918 Nov 18, 2021

github-actions bot added the doc-not-needed Your PR changes do not impact docs label Nov 18, 2021

Jason918 force-pushed the fix-blocking-ops-in-topic-creation branch from 9fad6cc to 1adf3fc Compare November 18, 2021 13:57

codelipenghui added this to the 2.10.0 milestone Nov 22, 2021

codelipenghui requested review from eolivelli, hangc0276, codelipenghui, 315157973 and merlimat November 22, 2021 13:52

Jason918 force-pushed the fix-blocking-ops-in-topic-creation branch from a2bbd56 to b918092 Compare November 22, 2021 14:59

Jason918 closed this Nov 23, 2021

Jason918 reopened this Nov 23, 2021

michaeljmarshall reviewed Nov 23, 2021

View reviewed changes

eolivelli requested changes Nov 23, 2021

View reviewed changes

Jason918 requested a review from eolivelli November 24, 2021 06:06

Jason918 force-pushed the fix-blocking-ops-in-topic-creation branch from da61c1b to e1196d1 Compare November 30, 2021 06:47

codelipenghui reviewed Jan 18, 2022

View reviewed changes

Jason918 added 4 commits January 18, 2022 17:09

Fix blocking check backlog quota in Producer Creation

5f3c186

handle unexpected exceptions

978c093

remove isTimeBacklogExceeded

29c1c24

move exception handling outside

1b8ea00

Jason918 force-pushed the fix-blocking-ops-in-topic-creation branch from e1196d1 to 1b8ea00 Compare January 18, 2022 09:10

codelipenghui approved these changes Jan 21, 2022

View reviewed changes

jiazhai approved these changes Jan 21, 2022

View reviewed changes

mattisonchao approved these changes Jan 21, 2022

View reviewed changes

eolivelli approved these changes Jan 25, 2022

View reviewed changes

eolivelli merged commit 8c8738f into apache:master Jan 25, 2022

michaeljmarshall mentioned this pull request Jan 25, 2022

[ServerCnx] Only reply to client when code completes producerFuture #13949

Merged

1 task

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[broker] Optimize blocking backlogQuotaCheck to non-blocking in ServerCnx#handleProducer #12874

[broker] Optimize blocking backlogQuotaCheck to non-blocking in ServerCnx#handleProducer #12874

Jason918 commented Nov 18, 2021 •

edited

Loading

Jason918 commented Nov 19, 2021

Jason918 commented Nov 19, 2021

Jason918 commented Nov 19, 2021

Jason918 commented Nov 19, 2021

Jason918 commented Nov 19, 2021

Jason918 commented Nov 20, 2021

codelipenghui commented Nov 22, 2021

Jason918 commented Nov 22, 2021

Jason918 commented Nov 23, 2021

michaeljmarshall left a comment

eolivelli Nov 23, 2021

Jason918 Nov 24, 2021

eolivelli Nov 23, 2021

Jason918 Nov 24, 2021

eolivelli Nov 23, 2021

Jason918 Nov 24, 2021

eolivelli Nov 23, 2021

Jason918 Nov 24, 2021

merlimat Nov 29, 2021

Jason918 Nov 30, 2021

Jason918 commented Nov 24, 2021

Jason918 commented Nov 25, 2021

Jason918 commented Dec 6, 2021

codelipenghui Jan 18, 2022

Jason918 Jan 18, 2022

codelipenghui commented Jan 18, 2022

Jason918 commented Jan 18, 2022

codelipenghui commented Jan 21, 2022

eolivelli left a comment


		disableTcpNoDelayIfNeeded(topicName.toString(), producerName);
		backlogQuotaCheckFuture.thenRun(() -> {

[broker] Optimize blocking backlogQuotaCheck to non-blocking in ServerCnx#handleProducer #12874

[broker] Optimize blocking backlogQuotaCheck to non-blocking in ServerCnx#handleProducer #12874

Conversation

Jason918 commented Nov 18, 2021 • edited Loading

Motivation

Modifications

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Jason918 commented Nov 19, 2021

Jason918 commented Nov 19, 2021

Jason918 commented Nov 19, 2021

Jason918 commented Nov 19, 2021

Jason918 commented Nov 19, 2021

Jason918 commented Nov 20, 2021

codelipenghui commented Nov 22, 2021

Jason918 commented Nov 22, 2021

Jason918 commented Nov 23, 2021

michaeljmarshall 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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Jason918 commented Nov 24, 2021

Jason918 commented Nov 25, 2021

Jason918 commented Dec 6, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codelipenghui commented Jan 18, 2022

Jason918 commented Jan 18, 2022

codelipenghui commented Jan 21, 2022

eolivelli left a comment

Choose a reason for hiding this comment

Jason918 commented Nov 18, 2021 •

edited

Loading