Start namespace service and schema registry service before start broker. (#6499)

### Motivation

If the broker service is started, the client can connect to the broker and send requests depends on the namespace service, so we should create the namespace service before starting the broker. Otherwise, NPE occurs.

![image](https://user-images.githubusercontent.com/12592133/76090515-a9961400-5ff6-11ea-9077-cb8e79fa27c0.png)

![image](https://user-images.githubusercontent.com/12592133/76099838-b15db480-6006-11ea-8f39-31d820563c88.png)


### Modifications

Move the namespace service creation and the schema registry service creation before start broker service.

(cherry picked from commit 5285c68b)

pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java

@@ -401,6 +401,10 @@ public class PulsarService implements AutoCloseable {
             // Start load management service (even if load balancing is disabled)
             this.loadManager.set(LoadManager.create(this));
 
+            // needs load management service and before start broker service,
+            this.startNamespaceService();
+            schemaRegistryService = SchemaRegistryService.create(this);
+
             this.defaultOffloader = createManagedLedgerOffloader(
                     OffloadPolicies.create(this.getConfiguration().getProperties()));
 
@@ -458,8 +462,6 @@ public class PulsarService implements AutoCloseable {
             }
             this.webService.addStaticResources("/static", "/static");
 
-            schemaRegistryService = SchemaRegistryService.create(this);
-
             webService.start();
 
             // Refresh addresses, since the port might have been dynamically assigned
@@ -474,8 +476,8 @@ public class PulsarService implements AutoCloseable {
                 this.webSocketService.setLocalCluster(clusterData);
             }
 
-            // needs load management service
-            this.startNamespaceService();
+            // Initialize namespace service, after service url assigned. Should init zk and refresh self owner info.
+            this.nsService.initialize();
 
             // Start the leader election service
             startLeaderElectionService();

pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java

@@ -73,7 +73,6 @@ import org.slf4j.LoggerFactory;
 
 import java.net.URI;
 import java.net.URL;
-import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
@@ -158,13 +157,19 @@ public class NamespaceService {
         host = pulsar.getAdvertisedAddress();
         this.config = pulsar.getConfiguration();
         this.loadManager = pulsar.getLoadManager();
-        ServiceUnitZkUtils.initZK(pulsar.getLocalZkCache().getZooKeeper(), pulsar.getSafeBrokerServiceUrl());
         this.bundleFactory = new NamespaceBundleFactory(pulsar, Hashing.crc32());
         this.ownershipCache = new OwnershipCache(pulsar, bundleFactory, this);
         this.namespaceClients = new ConcurrentOpenHashMap<>();
         this.bundleOwnershipListeners = new CopyOnWriteArrayList<>();
     }
 
+    public void initialize() {
+        ServiceUnitZkUtils.initZK(pulsar.getLocalZkCache().getZooKeeper(), pulsar.getSafeBrokerServiceUrl());
+        if (!getOwnershipCache().refreshSelfOwnerInfo()) {
+            throw new RuntimeException("Failed to refresh self owner info.");
+        }
+    }
+
     public CompletableFuture<Optional<LookupResult>> getBrokerServiceUrlAsync(TopicName topic,
             boolean authoritative) {
         return getBundleAsync(topic)
@@ -424,7 +429,7 @@ public class NamespaceService {
         try {
             checkNotNull(candidateBroker);
 
-            if (pulsar.getSafeWebServiceAddress().equals(candidateBroker)) {
+            if (candidateBroker.equals(pulsar.getSafeWebServiceAddress())) {
                 // invalidate namespace policies and try to load latest policies to avoid data-discrepancy if broker
                 // doesn't receive watch on policies changes
                 final String policyPath = AdminResource.path(POLICIES, bundle.getNamespaceObject().toString());

pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/OwnershipCache.java

@@ -73,7 +73,7 @@ public class OwnershipCache {
     /**
      * The NamespaceEphemeralData objects that can be associated with the current owner
      */
-    private final NamespaceEphemeralData selfOwnerInfo;
+    private NamespaceEphemeralData selfOwnerInfo;
 
     /**
      * The NamespaceEphemeralData objects that can be associated with the current owner, when the broker is disabled.
@@ -111,6 +111,8 @@ public class OwnershipCache {
      */
     private NamespaceService namespaceService;
 
+    private final PulsarService pulsar;
+
     private class OwnedServiceUnitCacheLoader implements AsyncCacheLoader<String, OwnedBundle> {
 
         @SuppressWarnings("deprecation")
@@ -156,6 +158,7 @@ public class OwnershipCache {
      */
     public OwnershipCache(PulsarService pulsar, NamespaceBundleFactory bundleFactory, NamespaceService namespaceService) {
         this.namespaceService = namespaceService;
+        this.pulsar = pulsar;
         this.ownerBrokerUrl = pulsar.getSafeBrokerServiceUrl();
         this.ownerBrokerUrlTls = pulsar.getBrokerServiceUrlTls();
         this.selfOwnerInfo = new NamespaceEphemeralData(ownerBrokerUrl, ownerBrokerUrlTls,
@@ -211,6 +214,11 @@ public class OwnershipCache {
 
         CompletableFuture<NamespaceEphemeralData> future = new CompletableFuture<>();
 
+        if (!refreshSelfOwnerInfo()) {
+            future.completeExceptionally(new RuntimeException("Namespace service does not ready for acquiring ownership"));
+            return future;
+        }
+
         LOG.info("Trying to acquire ownership of {}", bundle);
 
         // Doing a get() on the ownedBundlesCache will trigger an async ZK write to acquire the lock over the
@@ -367,4 +375,12 @@ public class OwnershipCache {
     public NamespaceEphemeralData getSelfOwnerInfo() {
         return selfOwnerInfo;
     }
+
+    public synchronized boolean refreshSelfOwnerInfo() {
+        if (selfOwnerInfo.getNativeUrl() == null) {
+            this.selfOwnerInfo = new NamespaceEphemeralData(pulsar.getSafeBrokerServiceUrl(), pulsar.getBrokerServiceUrlTls(),
+                    pulsar.getSafeWebServiceAddress(), pulsar.getWebServiceAddressTls(), false);
+        }
+        return selfOwnerInfo.getNativeUrl() != null;
+    }
 }