[fix][broker] Support loadBalancerSheddingIntervalMinutes dynamic con…

…figuration (apache#16408) (cherry picked from commit e4dcf5a) (cherry picked from commit 18cd7f4)
datastax · Sep 2, 2022 · 3275c82 · 3275c82
1 parent 013cb01
commit 3275c82
Show file tree

Hide file tree

Showing 4 changed files with 53 additions and 10 deletions.
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
@@ -211,7 +211,7 @@ public class PulsarService implements AutoCloseable, ShutdownService {
     private LedgerOffloaderStats offloaderStats;
     private Map<NamespaceName, LedgerOffloader> ledgerOffloaderMap = new ConcurrentHashMap<>();
     private ScheduledFuture<?> loadReportTask = null;
-    private ScheduledFuture<?> loadSheddingTask = null;
+    private LoadSheddingTask loadSheddingTask = null;
     private ScheduledFuture<?> loadResourceQuotaTask = null;
     private final AtomicReference<LoadManager> loadManager = new AtomicReference<>();
     private PulsarAdmin adminClient = null;
@@ -1006,20 +1006,17 @@ protected void startLeaderElectionService() {
                     if (state == LeaderElectionState.Leading) {
                         LOG.info("This broker was elected leader");
                         if (getConfiguration().isLoadBalancerEnabled()) {
-                            long loadSheddingInterval = TimeUnit.MINUTES
-                                    .toMillis(getConfiguration().getLoadBalancerSheddingIntervalMinutes());
                             long resourceQuotaUpdateInterval = TimeUnit.MINUTES
                                     .toMillis(getConfiguration().getLoadBalancerResourceQuotaUpdateIntervalMinutes());
 
                             if (loadSheddingTask != null) {
-                                loadSheddingTask.cancel(false);
+                                loadSheddingTask.cancel();
                             }
                             if (loadResourceQuotaTask != null) {
                                 loadResourceQuotaTask.cancel(false);
                             }
-                            loadSheddingTask = loadManagerExecutor.scheduleAtFixedRate(
-                                    new LoadSheddingTask(loadManager),
-                                    loadSheddingInterval, loadSheddingInterval, TimeUnit.MILLISECONDS);
+                            loadSheddingTask = new LoadSheddingTask(loadManager, loadManagerExecutor, config);
+                            loadSheddingTask.start();
                             loadResourceQuotaTask = loadManagerExecutor.scheduleAtFixedRate(
                                     new LoadResourceQuotaUpdaterTask(loadManager), resourceQuotaUpdateInterval,
                                     resourceQuotaUpdateInterval, TimeUnit.MILLISECONDS);
@@ -1030,7 +1027,7 @@ protected void startLeaderElectionService() {
                                     leaderElectionService.getCurrentLeader());
                         }
                         if (loadSheddingTask != null) {
-                            loadSheddingTask.cancel(false);
+                            loadSheddingTask.cancel();
                             loadSheddingTask = null;
                         }
                         if (loadResourceQuotaTask != null) {

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LoadSheddingTask.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LoadSheddingTask.java
@@ -18,7 +18,10 @@
  */
 package org.apache.pulsar.broker.loadbalance;
 
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
+import org.apache.pulsar.broker.ServiceConfiguration;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -28,9 +31,18 @@
 public class LoadSheddingTask implements Runnable {
     private static final Logger LOG = LoggerFactory.getLogger(LoadSheddingTask.class);
     private final AtomicReference<LoadManager> loadManager;
+    private final ScheduledExecutorService loadManagerExecutor;
 
-    public LoadSheddingTask(AtomicReference<LoadManager> loadManager) {
+    private final ServiceConfiguration config;
+
+    private volatile boolean isCancel = false;
+
+    public LoadSheddingTask(AtomicReference<LoadManager> loadManager,
+                            ScheduledExecutorService loadManagerExecutor,
+                            ServiceConfiguration config) {
         this.loadManager = loadManager;
+        this.loadManagerExecutor = loadManagerExecutor;
+        this.config = config;
     }
 
     @Override
@@ -39,6 +51,26 @@ public void run() {
             loadManager.get().doLoadShedding();
         } catch (Exception e) {
             LOG.warn("Error during the load shedding", e);
+        } finally {
+            if (!isCancel && loadManagerExecutor != null && config != null) {
+                loadManagerExecutor.schedule(
+                        new LoadSheddingTask(loadManager, loadManagerExecutor, config),
+                        config.getLoadBalancerSheddingIntervalMinutes(),
+                        TimeUnit.MINUTES);
+            }
         }
     }
+
+    public void start() {
+        if (loadManagerExecutor != null && config != null) {
+            loadManagerExecutor.schedule(
+                    new LoadSheddingTask(loadManager, loadManagerExecutor, config),
+                    config.getLoadBalancerSheddingIntervalMinutes(),
+                    TimeUnit.MINUTES);
+        }
+    }
+
+    public void cancel() {
+        isCancel = true;
+    }
 }
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java
@@ -517,6 +517,20 @@ public void testUpdateDynamicCacheConfigurationWithZkWatch() throws Exception {
                 .toNanos(2000));
     }
 
+    @Test
+    public void testUpdateDynamicLoadBalancerSheddingIntervalMinutes() throws Exception {
+        // update configuration
+        admin.brokers().updateDynamicConfiguration("loadBalancerSheddingIntervalMinutes", "10");
+
+        // wait config to be updated
+        Awaitility.await().until(() -> {
+            return conf.getLoadBalancerSheddingIntervalMinutes() == 10;
+        });
+
+        // verify value is updated
+        assertEquals(conf.getLoadBalancerSheddingIntervalMinutes(), 10);
+    }
+
     /**
      * <pre>
      * Verifies: zk-update configuration updates service-config

diff --git a/...-broker/src/test/java/org/apache/pulsar/broker/loadbalance/SimpleLoadManagerImplTest.java b/...-broker/src/test/java/org/apache/pulsar/broker/loadbalance/SimpleLoadManagerImplTest.java
@@ -453,7 +453,7 @@ public void testTask() throws Exception {
         task1.run();
         verify(loadManager, times(1)).writeResourceQuotasToZooKeeper();
 
-        LoadSheddingTask task2 = new LoadSheddingTask(atomicLoadManager);
+        LoadSheddingTask task2 = new LoadSheddingTask(atomicLoadManager, null, null);
         task2.run();
         verify(loadManager, times(1)).doLoadShedding();
     }