[pulsar] branch branch-3.0 updated: [fix][broker] getOwnedServiceUnits NPE (#20625)

[technoboy]

This is an automated email from the ASF dual-hosted git repository.

technoboy pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git


The following commit(s) were added to refs/heads/branch-3.0 by this push:
     new c3cae36a6ff [fix][broker] getOwnedServiceUnits NPE (#20625)
c3cae36a6ff is described below

commit c3cae36a6ff8a378e622da49c1d2453fc0e05d25
Author: Kai Wang <kw...@apache.org>
AuthorDate: Wed Jun 21 13:27:16 2023 +0800

    [fix][broker] getOwnedServiceUnits NPE (#20625)
---
 .../broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java  | 6 +++++-
 .../loadbalance/extensions/ExtensibleLoadManagerImplTest.java     | 8 ++++++++
 2 files changed, 13 insertions(+), 1 deletion(-)

diff --git 
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java
 
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java
index a9a6e6138bd..d0c55a65191 100644
--- 
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java
+++ 
b/pulsar-broker/src/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/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java
 
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java
index dc844a3d031..7af10c4aadc 100644
--- 
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java
+++ 
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java
@@ -1115,6 +1115,14 @@ public class ExtensibleLoadManagerImplTest extends 
MockedPulsarServiceBaseTest {
         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 {