[fix][broker] getOwnedServiceUnits NPE (#20625)

apache · Jun 21, 2023 · 7bb3531 · 7bb3531
1 parent ffc84c9
commit 7bb3531
Show file tree

Hide file tree

Showing 2 changed files with 13 additions and 1 deletion.
diff --git a/.../main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java b/.../main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java
@@ -169,7 +169,7 @@ public class ExtensibleLoadManagerImpl implements ExtensibleLoadManager {
 
     private SplitManager splitManager;
 
-    private boolean started = false;
+    private volatile boolean started = false;
 
     private final AssignCounter assignCounter = new AssignCounter();
     @Getter
@@ -191,6 +191,10 @@ public class ExtensibleLoadManagerImpl implements ExtensibleLoadManager {
      * Get all the bundles that are owned by this broker.
      */
     public Set<NamespaceBundle> getOwnedServiceUnits() {
+        if (!started) {
+            log.warn("Failed to get owned service units, load manager is not started.");
+            return Collections.emptySet();
+        }
         Set<Map.Entry<String, ServiceUnitStateData>> entrySet = serviceUnitStateChannel.getOwnershipEntrySet();
         String brokerId = brokerRegistry.getBrokerId();
         return entrySet.stream()

diff --git a/...t/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java b/...t/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java
@@ -1115,6 +1115,14 @@ private void assertOwnedServiceUnits(
         assertEquals(status.broker_assignment, BrokerAssignment.shared);
     }
 
+    @Test(timeOut = 30 * 1000)
+    public void testGetOwnedServiceUnitsWhenLoadManagerNotStart() {
+        ExtensibleLoadManagerImpl loadManager = new ExtensibleLoadManagerImpl();
+        Set<NamespaceBundle> ownedServiceUnits = loadManager.getOwnedServiceUnits();
+        assertNotNull(ownedServiceUnits);
+        assertTrue(ownedServiceUnits.isEmpty());
+    }
+
     @Test(timeOut = 30 * 1000)
     public void testTryAcquiringOwnership()
             throws PulsarAdminException, ExecutionException, InterruptedException {