[fix][broker] Ensure LoadSheddingTask is scheduled after metadata service is available again (apache#24838)

Demogorgon314 · lhotari · commit 49d425793492 · 2025-10-14T15:44:40.000+03:00
Motivation After PR: apache#23040, if the metadata service is unavailable and then becomes available again, the LoadSheddingTask will not run again. Modifications Ensure LoadSheddingTask is scheduled after metadata service is available again by moving the isMetadataServiceAvailable check to the try block. (cherry picked from commit 461ffd8)
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
@@ -59,11 +59,11 @@ public void run() {
         if (isCancel) {
             return;
         }
-        if (factory instanceof ManagedLedgerFactoryImpl
-                && !((ManagedLedgerFactoryImpl) factory).isMetadataServiceAvailable()) {
-            return;
-        }
         try {
+            if (factory instanceof ManagedLedgerFactoryImpl
+                    && !((ManagedLedgerFactoryImpl) factory).isMetadataServiceAvailable()) {
+                return;
+            }
             loadManager.get().doLoadShedding();
         } catch (Exception e) {
             LOG.warn("Error during the load shedding", e);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/SimpleLoadManagerImplTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/SimpleLoadManagerImplTest.java
@@ -22,6 +22,7 @@
 import static org.mockito.Mockito.atLeastOnce;
 import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.testng.Assert.assertEquals;
@@ -521,12 +522,14 @@ public void testMetadataServiceNotAvailable() {
         AtomicReference<LoadManager> atomicLoadManager = new AtomicReference<>(loadManager);
         ManagedLedgerFactoryImpl factory = mock(ManagedLedgerFactoryImpl.class);
         doReturn(false).when(factory).isMetadataServiceAvailable();
-        LoadSheddingTask task2 = new LoadSheddingTask(atomicLoadManager, null, null, factory);
+        LoadSheddingTask task2 = spy(new LoadSheddingTask(atomicLoadManager, null, null, factory));
         task2.run();
         verify(loadManager, times(0)).doLoadShedding();
+        verify(task2, times(1)).start();
         doReturn(true).when(factory).isMetadataServiceAvailable();
         task2.run();
         verify(loadManager, times(1)).doLoadShedding();
+        verify(task2, times(2)).start();
     }
 
     @Test
