Fixed NPE and cache invalidation in leader election

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

@@ -727,8 +727,10 @@ protected void startLeaderElectionService() {
                                     resourceQuotaUpdateInterval, TimeUnit.MILLISECONDS);
                         }
                     } else {
-                        LOG.info("This broker is a follower. Current leader is {}",
-                                leaderElectionService.getCurrentLeader());
+                        if (leaderElectionService != null) {
+                            LOG.info("This broker is a follower. Current leader is {}",
+                                    leaderElectionService.getCurrentLeader());
+                        }
                         if (loadSheddingTask != null) {
                             loadSheddingTask.cancel(false);
                             loadSheddingTask = null;

pulsar-metadata/src/main/java/org/apache/pulsar/metadata/api/MetadataCache.java

@@ -130,4 +130,11 @@ public interface MetadataCache<T> {
      *             if the object is not present in the metadata store.
      */
     CompletableFuture<Void> delete(String path);
+
+    /**
+     * Force the invalidation of an object in the metadata cache.
+     *
+     * @param path the path of the object in the metadata store
+     */
+    void invalidate(String path);
 }

pulsar-metadata/src/main/java/org/apache/pulsar/metadata/cache/impl/MetadataCacheImpl.java

@@ -227,6 +227,7 @@ public CompletableFuture<List<String>> getChildren(String path) {
         return store.getChildren(path);
     }
 
+    @Override
     public void invalidate(String path) {
         objCache.synchronous().invalidate(path);
     }

pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/LeaderElectionImpl.java

@@ -171,6 +171,12 @@ private synchronized CompletableFuture<LeaderElectionState> tryToBecomeLeader()
                     if (ex.getCause() instanceof BadVersionException) {
                         // There was a conflict between 2 participants trying to become leaders at same time. Retry
                         // to fetch info on new leader.
+
+                        // We force the invalidation of the cache entry. Since we received a BadVersion error, we
+                        // already know that the entry is out of date. If we don't invalidate, we'd be retrying the
+                        // leader election many times until we finally receive the notification that invalidates the
+                        // cache.
+                        cache.invalidate(path);
                         elect()
                             .thenAccept(lse -> result.complete(lse))
                             .exceptionally(ex2 -> {
@@ -259,10 +265,12 @@ public void accept(Notification notification) {
                                         log.warn("Exception in state change listener", t);
                                     }
 
-                                    executor.schedule(() -> {
-                                        log.info("Retrying Leader election for path {}");
-                                        elect();
-                                    }, LEADER_ELECTION_RETRY_DELAY_SECONDS, TimeUnit.SECONDS);
+                                    if (internalState != InternalState.Closed) {
+                                        executor.schedule(() -> {
+                                            log.info("Retrying Leader election for path {}");
+                                            elect();
+                                        }, LEADER_ELECTION_RETRY_DELAY_SECONDS, TimeUnit.SECONDS);
+                                    }
                                 }
                                 return null;
                             });