Change to Optional.empty() back

Author: Demogorgon314

pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java

@@ -56,9 +56,9 @@ public class ExtensibleLoadManagerImpl implements ExtensibleLoadManager {
     private final BrokerSelectionStrategy brokerSelectionStrategy;
 
     @Getter
-    private List<BrokerFilter> brokerFilterPipeline;
+    private final List<BrokerFilter> brokerFilterPipeline;
 
-    private volatile boolean started = false;
+    private boolean started = false;
 
     private final ConcurrentOpenHashMap<String, CompletableFuture<Optional<BrokerLookupData>>>
             lookupRequests = ConcurrentOpenHashMap.<String,
@@ -107,7 +107,7 @@ public synchronized void start() throws PulsarServerException {
     }
 
     @Override
-    public void initialize(PulsarService pulsar) {
+    public synchronized void initialize(PulsarService pulsar) {
         this.pulsar = pulsar;
         this.conf = pulsar.getConfiguration();
     }

pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java

@@ -246,10 +246,7 @@ public CompletableFuture<Optional<String>> getChannelOwnerAsync() {
                         broker = broker.substring(broker.lastIndexOf('/') + 1);
                         return Optional.of(broker);
                     } else {
-                        // When leader is empty, we should throw exception to notify is failed.
-                        String msg = "There is no channel owner now.";
-                        log.error(msg);
-                        throw new IllegalStateException(msg);
+                        return Optional.empty();
                     }
                 }
         );