Add topic loading throttling #306

rdhabalia · 2017-03-22T17:50:07Z

Motivation

Addressing #300 : At the time of broker cold-restart, clients make broker to load all the topics concurrently which can create back pressure at zk and may increase zk latency. So, broker should have a way to throttle number of concurrent topic loading.

Modifications

Introduce dynamic throttling limit to control concurrent topic loading.
above that limit broker rejects producer/consumer create request so, producer and consumer can do retry with backoff

Result

It will help broker to control number of zk operations on cold-restart.

merlimat

The change looks good. I just have 1 small concern.

The topic loading is initiated for all the topics in a given bundle/namespace, even though the particular topic is not being used. The main reason was to ensure stats visibility of that topic.

If we reject the loading, an unused topic will not get loaded later. That's why I was also thinking of "queuing" the topic load operations.

merlimat · 2017-03-22T17:57:35Z

pulsar-broker/src/main/java/com/yahoo/pulsar/broker/service/BrokerService.java

@@ -458,20 +467,23 @@ public void openLedgerComplete(ManagedLedger ledger, Object ctx) {

                                return null;
                            });
+                            topicLoadRequestSemaphore.get().release();


The acquire() and release() operations might not happen on the same semaphore if the limit is being changed.

You should save the instance into a local variable and call release on the same object.

yes, that make sense. updated it.

…e configuration

rdhabalia · 2017-03-22T20:43:56Z

If we reject the loading, an unused topic will not get loaded later. That's why I was also thinking of "queuing" the topic load operations.

That's correct. However, if topic is unused (no producer/consumer) then isn't it deleted by GC from that broker. So, after that it will not reflect into stats. And if there is any producer wants to connect then anyway that topic will be loaded after sometime.
Is that correct?

merlimat · 2017-03-22T20:44:49Z

However, if topic is unused (no producer/consumer) then isn't it deleted by GC from that broker.

The topic can have subscriptions and no producers/consumers using it

rdhabalia · 2017-03-22T23:23:50Z

The topic can have subscriptions and no producers/consumers using it

That's true, GC doesn't delete topic with non-empty subscription so, we can get the stat. Updated the change.

merlimat · 2017-03-23T18:07:53Z

pulsar-broker/src/main/java/com/yahoo/pulsar/broker/service/BrokerService.java

+
+        if (!topicLoadSemaphore.tryAcquire()) {
+            if (failOnTooManyConcurrentTopics) {
+                throw new TooManyRequestsException("Too many concurrent topic load exception");


Couldn't we just return a failed future at this point?

the main idea behind failing immediately with exception so, while retrying create topic from pendingTopicLoadingQueue if it fails again here then we can put same topic-future again into the queue because it has original future which we had returned back to ServerCnx on very first time when createPersistentTopic request came.

merlimat · 2017-03-23T18:08:51Z

pulsar-broker/src/main/java/com/yahoo/pulsar/broker/service/BrokerService.java

@@ -418,6 +431,15 @@ public PulsarClient getReplicationClient(String cluster) {
    }

    private CompletableFuture<Topic> createPersistentTopic(final String topic) throws RuntimeException {
+        try {
+            return createPersistentTopic(topic, false);


It is not immediate to understand what false means here

yes, we can add comment. But the idea of having this flag:

when BrokerService receives createPersistentTopic request, passing failOnTooManyConcurrentTopics=false means we will not fail the request due to throttling but will add into the queue.

But when we retry from pendingTopicLoadingQueue and again if it gets throttle then we add that topic-future again into the queue at line1037

merlimat · 2017-03-23T18:10:43Z

pulsar-broker/src/main/java/com/yahoo/pulsar/broker/service/BrokerService.java

@@ -870,7 +913,10 @@ private void updateConfigurationAndRegisterListeners() {
        updateDynamicServiceConfiguration();
        // add listener on "maxConcurrentLookupRequest" value change
        registerConfigurationListener("maxConcurrentLookupRequest",
-                (pendingLookupRequest) -> lookupRequestSemaphore.set(new Semaphore((int) pendingLookupRequest, true)));
+                (maxConcurrentLookupRequest) -> lookupRequestSemaphore.set(new Semaphore((int) maxConcurrentLookupRequest, true)));


I think we shouldn't strictly require the "fairness" in the semaphore.

you means we should not create a new semaphore? I have set the new instance rather updating permit into existing semaphore so, we don't mess up with new permits (reduce/increment) number as lookupRequestSemaphore is already concurrently used by multiple threads.

No no, I was referring to the "true" as 2nd parameter for the semaphore constructor. That's about the fairness of the semaphore itself. I think we might not need the exact fairness, so we could relax that option.

merlimat · 2017-03-23T18:11:37Z

pulsar-broker/src/main/java/com/yahoo/pulsar/broker/service/BrokerService.java

@@ -972,4 +1018,31 @@ public void onUpdate(String path, Map<String, String> data, Stat stat) {
        return dynamicConfigurationMap;
    }

-}
+    private void processPendingLoadTopic() {
+        lock.lock();


Why do we need this lock? Would it be enough to poll() one item out of the queue and process it?

it is to prevent race condition which may cause topic reside in pendingTopicLoadingQueue forever.
eg:

if MaxConcurrentTopicLoadRequest=1 and if two topics (topic1, topic2) come at same time then one of them (topic2) will be added into pendingQueue.

as soon as topic1 completes: it tries to release semaphore and poll topic2 from queue to createTopic again, but if 3rd new topic3 acquired a permit before it and when it releases it may find pendingTopicLoadingQueue empty as we have already poll topic2 it before. after that topic2 will be added again on line-1037.

So, added lock as pendingTopicLoadingQueue.poll() and pendingTopicLoadingQueue.add(pendingTopic) is not atomic.
Do you think we should ignore this situation?

merlimat · 2017-03-23T18:11:59Z

pulsar-broker/src/main/java/com/yahoo/pulsar/broker/service/BrokerService.java

+            Pair<String, CompletableFuture<Topic>> pendingTopic = pendingTopicLoadingQueue.poll();
+            if (pendingTopic != null) {
+                try {
+                    createPersistentTopic(pendingTopic.getLeft(), true).handle((topic, ex) -> {


We should check if we still own the topic

merlimat · 2017-03-23T18:17:54Z

pulsar-broker/src/main/java/com/yahoo/pulsar/broker/service/BrokerService.java

+        }
+    }
+
+    private CompletableFuture<Topic> createPersistentTopic(final String topic, boolean failOnTooManyConcurrentTopics)


Instead of having these 2 methods, I was thinking something like:

CompletableFuture<Topic> createPersistentTopic(final String topic) { // Creates future and call method below or queue the request } void createPersistentTopic(String topic, CompletableFuture<Topic> topicFuture) { // To be called when we are ready to load the topic }

void createPersistentTopic(String topic, CompletableFuture<Topic> topicFuture) { // To be called when we are ready to load the topic }

Does this method throttle again and puts in the queue again or it just fails the future if semaphore doesn't have any additional permits?

@merlimat I think I updated changes as per your suggestion

merlimat · 2017-03-23T18:18:59Z

pulsar-broker/src/main/java/com/yahoo/pulsar/broker/service/BrokerService.java

+                        }
+                        return null;
+                    });
+                } catch (TooManyRequestsException te) {


We shouldn't even have to check again at this point.

what if this time(as mentioned in above comment) some other new topic3 (from different bundle) comes and acquired a permit and there could be possibility that creation of this topic may fail again with TooManyRequestsException. So, we have to again put it back to the queue, as we don't want to fail it and try again.
So, shouldn't we handle retry-failure and put it back to the queue again?

saandrews · 2017-03-23T23:19:21Z

pulsar-broker/src/test/java/com/yahoo/pulsar/client/impl/BrokerClientIntegrationTest.java

+        latch.await();
+        FutureUtil.waitForAll(futures).get();
+
+        pulsarClient.close();


Can you add test to verify negative scenario with TooManyReqeusts exception?

actually as we decided broker should not give TooManyReqeustsException to client and put the topic in to the queue so, it give chance to unused topic to be loaded for stats visibility. So, eventually all the producers should get connected even it gets error internally at broker.

merlimat · 2017-03-24T16:11:08Z

pulsar-broker/src/main/java/com/yahoo/pulsar/broker/service/BrokerService.java

+        final Semaphore topicLoadSemaphore = topicLoadRequestSemaphore.get();
+
+        if (topicLoadSemaphore.tryAcquire()) {
+            CompletableFuture<Topic> future = new CompletableFuture<>();


I think this could be done with just 1 future, just the same way you add the handle method to attach behavior after completing

merlimat · 2017-03-24T16:13:34Z

pulsar-broker/src/main/java/com/yahoo/pulsar/broker/service/BrokerService.java

@@ -870,7 +913,10 @@ private void updateConfigurationAndRegisterListeners() {
        updateDynamicServiceConfiguration();
        // add listener on "maxConcurrentLookupRequest" value change
        registerConfigurationListener("maxConcurrentLookupRequest",
-                (pendingLookupRequest) -> lookupRequestSemaphore.set(new Semaphore((int) pendingLookupRequest, true)));
+                (maxConcurrentLookupRequest) -> lookupRequestSemaphore.set(new Semaphore((int) maxConcurrentLookupRequest, true)));


No no, I was referring to the "true" as 2nd parameter for the semaphore constructor. That's about the fairness of the semaphore itself. I think we might not need the exact fairness, so we could relax that option.

merlimat · 2017-03-24T16:15:28Z

pulsar-broker/src/main/java/com/yahoo/pulsar/broker/service/BrokerService.java

+     */
+    private void createPendingLoadTopic() {
+        Pair<String, CompletableFuture<Topic>> pendingTopic = pendingTopicLoadingQueue.poll();
+        if (pendingTopic != null) {


if (pendingtopic == null) return;

merlimat · 2017-03-24T16:16:58Z

pulsar-broker/src/main/java/com/yahoo/pulsar/broker/service/BrokerService.java

+                checkTopicNsOwnership(topic);
+                CompletableFuture<Topic> future = new CompletableFuture<>();
+                final Semaphore topicLoadSemaphore = topicLoadRequestSemaphore.get();
+                final boolean acquiredPermit = topicLoadSemaphore.tryAcquire();


What is the reason for acquiring the semaphore here, if we're not waiting for it anyway?

actually with this approach compare to my previous commit-406b0fc fails to provide throttling in certain usecases.
eg:

maxConcurrentTopicLoadRequest=10 and broker receives 21 requests so, 10 will be executed and rest of them will go to pending-queue

once first 10 completes it will try to execute pending-10 without any throttling. but at the same time if additional 31 requests come then 10 requests out of them will be served as semaphore is already released. so, now we are serving total 20 concurrent request (10(new requests) + 10(10 completed topic threads which invoked createPendingLoadTopic)

and this process go on which can increase concurrent-count and we are not doing real throttling.

However, I am acquiring semaphore here so, at least those new additional coming 31 requests will be throttled. Still this approach doesn't throttle 100% in compare to commit-406b0fc

Ok, that makes sense

merlimat · 2017-03-24T16:17:43Z

pulsar-broker/src/main/java/com/yahoo/pulsar/broker/service/BrokerService.java

+            final String topic = pendingTopic.getLeft();
+            try {
+                checkTopicNsOwnership(topic);
+                CompletableFuture<Topic> future = new CompletableFuture<>();


Same as above, problably we don't need an additional future

merlimat · 2017-03-24T16:18:12Z

pulsar-broker/src/main/java/com/yahoo/pulsar/broker/service/BrokerService.java

+                log.error("Failed to create pending topic {} {}", topic, re);
+                pendingTopic.getRight().completeExceptionally(re.getCause());
+                // process next pending topic
+                createPendingLoadTopic();


This would be a recursive call

createPersistent is async call so, we are making it recursive once future is completed otherwise due to throttling-limit=10 if queue has 1000 pending requests then all 1000 can go concurrently through createPersistent(topic,future) . So, we have to call it recursively only after receiving completed future. right?

But in this case if you get an exception when starting the create topic asynchronous, t will immediately start the next one. If all th queued topics start failing we'd likely get into stack overflow errors. Easy fix would be to schedule the next topic load from executor

that make sense.

merlimat

👍

merlimat · 2017-03-24T19:31:09Z

Couple of seemingly legitimate unit test failures:

ServerCnxTest.testProducerOnNotOwnedTopic:397 expected [ServiceNotReady] but found [UnknownError]
  BrokerClientIntegrationTest.testMaxConcurrentTopicLoading:579 » Execution com....

rdhabalia · 2017-03-24T22:25:23Z

yes, fixed.

Add topic loading throttling

d952ed9

rdhabalia added the type/enhancement The enhancements for the existing features or docs. e.g. reduce memory usage of the delayed messages label Mar 22, 2017

rdhabalia added this to the 1.17 milestone Mar 22, 2017

rdhabalia self-assigned this Mar 22, 2017

merlimat reviewed Mar 22, 2017

View reviewed changes

local semaphore instance to avoid incorrect semaphore access on updat…

2205792

…e configuration

add topic-loading into pending list on tooManyRequest

406b0fc

rdhabalia force-pushed the topic_throttle branch from 30d1193 to 406b0fc Compare March 22, 2017 23:22

merlimat reviewed Mar 23, 2017

View reviewed changes

saandrews reviewed Mar 23, 2017

View reviewed changes

merlimat reviewed Mar 24, 2017

View reviewed changes

rdhabalia force-pushed the topic_throttle branch 3 times, most recently from 78b2aca to 8027740 Compare March 24, 2017 18:44

merlimat approved these changes Mar 24, 2017

View reviewed changes

process pendingTopicRequest iteratively without throwing exception

598954d

rdhabalia force-pushed the topic_throttle branch from 8027740 to 598954d Compare March 24, 2017 21:13

merlimat merged commit 2886250 into apache:master Mar 24, 2017

rdhabalia deleted the topic_throttle branch June 21, 2017 18:54

xiaotongwang1 mentioned this pull request Aug 4, 2021

Pulsar 2.7.0+ KOP 2.7.2.x getPartitionedTopicMetadata timeout #11532

Closed

Add topic loading throttling #306

Add topic loading throttling #306

Conversation

rdhabalia commented Mar 22, 2017

Motivation

Modifications

Result

merlimat left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rdhabalia commented Mar 22, 2017

merlimat commented Mar 22, 2017

rdhabalia commented Mar 22, 2017

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rdhabalia Mar 23, 2017 • edited Loading

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

merlimat left a comment

Choose a reason for hiding this comment

merlimat commented Mar 24, 2017

rdhabalia commented Mar 24, 2017

rdhabalia Mar 23, 2017 •

edited

Loading