Configure split-bundle task interval different than loadReportGenerationTask-interval (#183)

conf/broker.conf

@@ -237,9 +237,6 @@ loadBalancerBrokerComfortLoadLevelPercentage=65
 # enable/disable namespace bundle auto split
 loadBalancerAutoBundleSplitEnabled=false
 
-# interval to detect & split hot namespace bundle
-loadBalancerNamespaceBundleSplitIntervalMinutes=15
-
 # maximum topics in a bundle, otherwise bundle split will be triggered
 loadBalancerNamespaceBundleMaxTopics=1000
 

conf/standalone.conf

@@ -210,9 +210,6 @@ loadBalancerBrokerComfortLoadLevelPercentage=65
 # enable/disable namespace bundle auto split
 loadBalancerAutoBundleSplitEnabled=false
 
-# interval to detect & split hot namespace bundle
-loadBalancerNamespaceBundleSplitIntervalMinutes=15
-
 # maximum topics in a bundle, otherwise bundle split will be triggered
 loadBalancerNamespaceBundleMaxTopics=1000
 

pulsar-broker-common/src/main/java/com/yahoo/pulsar/broker/ServiceConfiguration.java

@@ -207,8 +207,6 @@ public class ServiceConfiguration implements PulsarConfiguration{
     private int loadBalancerBrokerComfortLoadLevelPercentage = 65;
     // enable/disable automatic namespace bundle split
     private boolean loadBalancerAutoBundleSplitEnabled = false;
-    // interval to detect & split hot namespace bundle
-    private int loadBalancerNamespaceBundleSplitIntervalMinutes = 15;
     // maximum topics in a bundle, otherwise bundle split will be triggered
     private int loadBalancerNamespaceBundleMaxTopics = 1000;
     // maximum sessions (producers + consumers) in a bundle, otherwise bundle split will be triggered
@@ -759,14 +757,6 @@ public class ServiceConfiguration implements PulsarConfiguration{
         return loadBalancerBrokerOverloadedThresholdPercentage;
     }
 
-    public void setLoadBalancerBundleSplitIntervalMinutes(int interval) {
-        this.loadBalancerNamespaceBundleSplitIntervalMinutes = interval;
-    }
-
-    public int getLoadBalancerBundleSplitIntervalMinutes() {
-        return this.loadBalancerNamespaceBundleSplitIntervalMinutes;
-    }
-
     public void setLoadBalancerNamespaceBundleMaxTopics(int topics) {
         this.loadBalancerNamespaceBundleMaxTopics = topics;
     }

pulsar-broker/src/main/java/com/yahoo/pulsar/broker/PulsarService.java

@@ -45,7 +45,6 @@ import com.yahoo.pulsar.broker.loadbalance.LoadManager;
 import com.yahoo.pulsar.broker.loadbalance.LoadReportUpdaterTask;
 import com.yahoo.pulsar.broker.loadbalance.LoadResourceQuotaUpdaterTask;
 import com.yahoo.pulsar.broker.loadbalance.LoadSheddingTask;
-import com.yahoo.pulsar.broker.loadbalance.NamespaceBundleSplitTask;
 import com.yahoo.pulsar.broker.loadbalance.impl.SimpleLoadManagerImpl;
 import com.yahoo.pulsar.broker.namespace.NamespaceService;
 import com.yahoo.pulsar.broker.service.BrokerService;
@@ -92,7 +91,6 @@ public class PulsarService implements AutoCloseable {
     private final OrderedSafeExecutor orderedExecutor = new OrderedSafeExecutor(8, "pulsar-ordered");
     private ScheduledExecutorService loadManagerExecutor = null;
     private ScheduledFuture<?> loadReportTask = null;
-    private ScheduledFuture<?> bundleSplitTask = null;
     private ScheduledFuture<?> loadSheddingTask = null;
     private ScheduledFuture<?> loadResourceQuotaTask = null;
     private LoadManager loadManager = null;
@@ -408,13 +406,6 @@ public class PulsarService implements AutoCloseable {
                         new LoadReportUpdaterTask(loadManager), loadReportMinInterval, loadReportMinInterval,
                         TimeUnit.MILLISECONDS);
             }
-            if (this.bundleSplitTask == null) {
-                long bundleSplitInterval = TimeUnit.MINUTES
-                        .toMillis(getConfiguration().getLoadBalancerBundleSplitIntervalMinutes());
-                this.bundleSplitTask = this.loadManagerExecutor.scheduleAtFixedRate(
-                        new NamespaceBundleSplitTask(loadManager), bundleSplitInterval, bundleSplitInterval,
-                        TimeUnit.MILLISECONDS);
-            }
         }
     }
 

pulsar-broker/src/main/java/com/yahoo/pulsar/broker/loadbalance/NamespaceBundleSplitTask.java

-/**
- * Copyright 2016 Yahoo Inc.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package com.yahoo.pulsar.broker.loadbalance;
-
-import java.lang.Runnable;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Represents a task which detect hot namespace bundle to split
- */
-public class NamespaceBundleSplitTask implements Runnable {
-    private static final Logger LOG = LoggerFactory.getLogger(NamespaceBundleSplitTask.class);
-    private final LoadManager loadManager;
-
-    public NamespaceBundleSplitTask(LoadManager manager) {
-        loadManager = manager;
-    }
-
-    @Override
-    public void run() {
-        try {
-            loadManager.doNamespaceBundleSplit();
-        } catch (Exception e) {
-            LOG.warn("Error while doing namespace bundle split.", e);
-        }
-    }
-}

pulsar-broker/src/main/java/com/yahoo/pulsar/broker/loadbalance/impl/SimpleLoadManagerImpl.java

@@ -1199,6 +1199,8 @@ public class SimpleLoadManagerImpl implements LoadManager, ZooKeeperCacheListene
                 -1);
             this.lastLoadReport = lr;
             this.lastResourceUsageTimestamp = lr.getTimestamp();
+            // split-bundle if requires
+            doNamespaceBundleSplit();
         }
     }
 

pulsar-broker/src/test/java/com/yahoo/pulsar/broker/loadbalance/SimpleLoadManagerImplTest.java

@@ -459,10 +459,6 @@ public class SimpleLoadManagerImplTest {
         LoadSheddingTask task2 = new LoadSheddingTask(loadManager);
         task2.run();
         verify(loadManager, times(1)).doLoadShedding();
-
-        NamespaceBundleSplitTask task3 = new NamespaceBundleSplitTask(loadManager);
-        task3.run();
-        verify(loadManager, times(1)).doNamespaceBundleSplit();
     }
 
     @Test