[cherry-pick][branch-2.9] Fix issue where leader broker information isn't available after 10 minutes

pulsar-broker/src/test/java/org/apache/pulsar/broker/MultiBrokerTestZKBaseTest.java

@@ -0,0 +1,61 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.metadata.TestZKServer;
+import org.apache.pulsar.metadata.api.MetadataStoreConfig;
+import org.apache.pulsar.metadata.api.MetadataStoreException;
+import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended;
+
+/**
+ * Multiple brokers with a real test Zookeeper server (instead of the mock server)
+ */
+@Slf4j
+public abstract class MultiBrokerTestZKBaseTest extends MultiBrokerBaseTest {
+    TestZKServer testZKServer;
+
+    @Override
+    protected void doInitConf() throws Exception {
+        super.doInitConf();
+        testZKServer = new TestZKServer();
+    }
+
+    @Override
+    protected void onCleanup() {
+        super.onCleanup();
+        if (testZKServer != null) {
+            try {
+                testZKServer.close();
+            } catch (Exception e) {
+                log.error("Error in stopping ZK server", e);
+            }
+        }
+    }
+
+    @Override
+    protected MetadataStoreExtended createLocalMetadataStore() throws MetadataStoreException {
+        return MetadataStoreExtended.create(testZKServer.getConnectionString(), MetadataStoreConfig.builder().build());
+    }
+
+    @Override
+    protected MetadataStoreExtended createConfigurationMetadataStore() throws MetadataStoreException {
+        return MetadataStoreExtended.create(testZKServer.getConnectionString(), MetadataStoreConfig.builder().build());
+    }
+}

pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/MultiBrokerLeaderElectionExpirationTest.java

@@ -0,0 +1,114 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.pulsar.broker.loadbalance;
+
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+import java.util.Optional;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.broker.MultiBrokerTestZKBaseTest;
+import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.metadata.api.MetadataCacheConfig;
+import org.apache.pulsar.metadata.api.MetadataStoreException;
+import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended;
+import org.awaitility.Awaitility;
+import org.testng.annotations.Test;
+
+@Slf4j
+@Test(groups = "broker")
+public class MultiBrokerLeaderElectionExpirationTest extends MultiBrokerTestZKBaseTest {
+    private static final long EXPIRE_AFTER_WRITE_MILLIS_IN_TEST = 2000L;
+    private static final long REFRESH_AFTER_WRITE_MILLIS_IN_TEST = 1000L;
+
+    @Override
+    protected int numberOfAdditionalBrokers() {
+        return 9;
+    }
+
+    @Test
+    public void shouldElectOneLeader() {
+        int leaders = 0;
+        for (PulsarService broker : getAllBrokers()) {
+            if (broker.getLeaderElectionService().isLeader()) {
+                leaders++;
+            }
+        }
+        assertEquals(leaders, 1);
+    }
+
+    @Override
+    protected MetadataStoreExtended createLocalMetadataStore() throws MetadataStoreException {
+        return changeDefaultMetadataCacheConfig(super.createLocalMetadataStore());
+    }
+
+    @Override
+    protected MetadataStoreExtended createConfigurationMetadataStore() throws MetadataStoreException {
+        return changeDefaultMetadataCacheConfig(super.createConfigurationMetadataStore());
+    }
+
+    MetadataStoreExtended changeDefaultMetadataCacheConfig(MetadataStoreExtended metadataStore) {
+        MetadataStoreExtended spy = spy(metadataStore);
+        when(spy.getDefaultMetadataCacheConfig()).thenReturn(MetadataCacheConfig
+                .builder()
+                .refreshAfterWriteMillis(REFRESH_AFTER_WRITE_MILLIS_IN_TEST)
+                .expireAfterWriteMillis(EXPIRE_AFTER_WRITE_MILLIS_IN_TEST)
+                .build());
+        return spy;
+    }
+
+    @Test
+    public void shouldAllBrokersBeAbleToGetTheLeaderAfterExpiration()
+            throws ExecutionException, InterruptedException, TimeoutException {
+
+        // if you want to see this test fail, modify the line in LeaderElectionImpl constructor for creating
+        // the metadata cache to not skip expirations:
+        // this.cache = store.getMetadataCache(clazz);
+
+        // Given that all brokers have the leader elected
+        Awaitility.await().untilAsserted(() -> {
+            for (PulsarService broker : getAllBrokers()) {
+                Optional<LeaderBroker> currentLeader = broker.getLeaderElectionService().getCurrentLeader();
+                assertTrue(currentLeader.isPresent(), "Leader wasn't known on broker " + broker.getBrokerServiceUrl());
+            }
+        });
+
+        // Wait for metadata cache entries to expire
+        Thread.sleep(EXPIRE_AFTER_WRITE_MILLIS_IN_TEST);
+
+        // then leader should be known on all brokers and it should be the same leader
+        LeaderBroker leader = null;
+        for (PulsarService broker : getAllBrokers()) {
+            Optional<LeaderBroker> currentLeader =
+                    broker.getLeaderElectionService().readCurrentLeader().get(1, TimeUnit.SECONDS);
+            assertTrue(currentLeader.isPresent(), "Leader wasn't known on broker " + broker.getBrokerServiceUrl());
+            if (leader != null) {
+                assertEquals(currentLeader.get(), leader,
+                        "Different leader on broker " + broker.getBrokerServiceUrl());
+            } else {
+                leader = currentLeader.get();
+            }
+        }
+    }
+}

pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/MultiBrokerLeaderElectionTest.java

@@ -34,55 +34,21 @@
 import java.util.stream.IntStream;
 import lombok.Cleanup;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.pulsar.broker.MultiBrokerBaseTest;
+import org.apache.pulsar.broker.MultiBrokerTestZKBaseTest;
 import org.apache.pulsar.broker.PulsarService;
 import org.apache.pulsar.client.admin.PulsarAdmin;
 import org.apache.pulsar.client.admin.PulsarAdminException;
-import org.apache.pulsar.metadata.TestZKServer;
-import org.apache.pulsar.metadata.api.MetadataStoreConfig;
-import org.apache.pulsar.metadata.api.MetadataStoreException;
-import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended;
 import org.awaitility.Awaitility;
 import org.testng.annotations.Test;
 
 @Slf4j
 @Test(groups = "broker")
-public class MultiBrokerLeaderElectionTest extends MultiBrokerBaseTest {
+public class MultiBrokerLeaderElectionTest extends MultiBrokerTestZKBaseTest {
     @Override
     protected int numberOfAdditionalBrokers() {
         return 9;
     }
 
-    TestZKServer testZKServer;
-
-    @Override
-    protected void doInitConf() throws Exception {
-        super.doInitConf();
-        testZKServer = new TestZKServer();
-    }
-
-    @Override
-    protected void onCleanup() {
-        super.onCleanup();
-        if (testZKServer != null) {
-            try {
-                testZKServer.close();
-            } catch (Exception e) {
-                log.error("Error in stopping ZK server", e);
-            }
-        }
-    }
-
-    @Override
-    protected MetadataStoreExtended createLocalMetadataStore() throws MetadataStoreException {
-        return MetadataStoreExtended.create(testZKServer.getConnectionString(), MetadataStoreConfig.builder().build());
-    }
-
-    @Override
-    protected MetadataStoreExtended createConfigurationMetadataStore() throws MetadataStoreException {
-        return MetadataStoreExtended.create(testZKServer.getConnectionString(), MetadataStoreConfig.builder().build());
-    }
-
     @Test
     public void shouldElectOneLeader() {
         int leaders = 0;

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

@@ -148,4 +148,11 @@ public interface MetadataCache<T> {
      * @param path the path of the object in the metadata store
      */
     void invalidate(String path);
+
+    /**
+     * Invalidate and reload an object in the metadata cache.
+     *
+     * @param path the path of the object in the metadata store
+     */
+    void refresh(String path);
 }

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

@@ -0,0 +1,50 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.metadata.api;
+
+import java.util.concurrent.TimeUnit;
+import lombok.Builder;
+import lombok.Getter;
+import lombok.ToString;
+
+/**
+ * The configuration builder for a {@link MetadataCache} config.
+ */
+@Builder
+@Getter
+@ToString
+public class MetadataCacheConfig {
+    private static final long DEFAULT_CACHE_REFRESH_TIME_MILLIS = TimeUnit.MINUTES.toMillis(5);
+
+    /**
+     * Specifies that active entries are eligible for automatic refresh once a fixed duration has
+     * elapsed after the entry's creation, or the most recent replacement of its value.
+     * A negative or zero value disables automatic refresh.
+     */
+    @Builder.Default
+    private final long refreshAfterWriteMillis = DEFAULT_CACHE_REFRESH_TIME_MILLIS;
+
+    /**
+     * Specifies that each entry should be automatically removed from the cache once a fixed duration
+     * has elapsed after the entry's creation, or the most recent replacement of its value.
+     * A negative or zero value disables automatic expiration.
+     */
+    @Builder.Default
+    private final long expireAfterWriteMillis = 2 * DEFAULT_CACHE_REFRESH_TIME_MILLIS;
+}

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

@@ -138,27 +138,78 @@ public interface MetadataStore extends AutoCloseable {
      * @param <T>
      * @param clazz
      *            the class type to be used for serialization/deserialization
+     * @param cacheConfig
+     *          the cache configuration to be used
      * @return the metadata cache object
      */
-    <T> MetadataCache<T> getMetadataCache(Class<T> clazz);
+    <T> MetadataCache<T> getMetadataCache(Class<T> clazz, MetadataCacheConfig cacheConfig);
+
+    /**
+     * Create a metadata cache specialized for a specific class.
+     *
+     * @param <T>
+     * @param clazz
+     *            the class type to be used for serialization/deserialization
+     * @return the metadata cache object
+     */
+    default <T> MetadataCache<T> getMetadataCache(Class<T> clazz) {
+        return getMetadataCache(clazz, getDefaultMetadataCacheConfig());
+    }
 
     /**
      * Create a metadata cache specialized for a specific class.
      *
      * @param <T>
      * @param typeRef
      *            the type ref description to be used for serialization/deserialization
+     * @param cacheConfig
+     *          the cache configuration to be used
      * @return the metadata cache object
      */
-    <T> MetadataCache<T> getMetadataCache(TypeReference<T> typeRef);
+    <T> MetadataCache<T> getMetadataCache(TypeReference<T> typeRef, MetadataCacheConfig cacheConfig);
+
+    /**
+     * Create a metadata cache specialized for a specific class.
+     *
+     * @param <T>
+     * @param typeRef
+     *            the type ref description to be used for serialization/deserialization
+     * @return the metadata cache object
+     */
+    default <T> MetadataCache<T> getMetadataCache(TypeReference<T> typeRef) {
+        return getMetadataCache(typeRef, getDefaultMetadataCacheConfig());
+    }
 
     /**
      * Create a metadata cache that uses a particular serde object.
      *
      * @param <T>
      * @param serde
      *            the custom serialization/deserialization object
+     * @param cacheConfig
+     *          the cache configuration to be used
      * @return the metadata cache object
      */
-    <T> MetadataCache<T> getMetadataCache(MetadataSerde<T> serde);
+    <T> MetadataCache<T> getMetadataCache(MetadataSerde<T> serde, MetadataCacheConfig cacheConfig);
+
+    /**
+     * Create a metadata cache that uses a particular serde object.
+     *
+     * @param <T>
+     * @param serde
+     *            the custom serialization/deserialization object
+     * @return the metadata cache object
+     */
+    default <T> MetadataCache<T> getMetadataCache(MetadataSerde<T> serde) {
+        return getMetadataCache(serde, getDefaultMetadataCacheConfig());
+    }
+
+    /**
+     * Returns the default metadata cache config.
+     *
+     * @return default metadata cache config
+     */
+    default MetadataCacheConfig getDefaultMetadataCacheConfig() {
+        return MetadataCacheConfig.builder().build();
+    }
 }