From 74901f7c7fd1589210c7ed98bc647b89701e8e5a Mon Sep 17 00:00:00 2001 From: Lari Hotari Date: Thu, 1 Sep 2022 15:08:05 +0300 Subject: [PATCH 1/3] [fix][broker] Leader election cache should not invalidate entries - cache expiration leads to problems and it is better to serve a stale entry than to ever expire the entry - MetadataStore didn't support passing cache configuration for a MetadataCache, it was necessary to modify the interface to support that. --- .../metadata/api/MetadataCacheConfig.java | 50 ++++++++++++++++ .../pulsar/metadata/api/MetadataStore.java | 57 ++++++++++++++++++- .../cache/impl/MetadataCacheImpl.java | 25 ++++---- .../coordination/impl/LeaderElectionImpl.java | 21 +++---- .../metadata/impl/AbstractMetadataStore.java | 13 +++-- .../impl/FaultInjectionMetadataStore.java | 13 +++-- 6 files changed, 143 insertions(+), 36 deletions(-) create mode 100644 pulsar-metadata/src/main/java/org/apache/pulsar/metadata/api/MetadataCacheConfig.java diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/api/MetadataCacheConfig.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/api/MetadataCacheConfig.java new file mode 100644 index 0000000000000..1671fd39956d8 --- /dev/null +++ b/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; +} diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/api/MetadataStore.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/api/MetadataStore.java index fa89d7e3dd24e..b4295f2586768 100644 --- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/api/MetadataStore.java +++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/api/MetadataStore.java @@ -136,9 +136,23 @@ public interface MetadataStore extends AutoCloseable { * @param * @param clazz * the class type to be used for serialization/deserialization + * @param cacheConfig + * the cache configuration to be used * @return the metadata cache object */ - MetadataCache getMetadataCache(Class clazz); + MetadataCache getMetadataCache(Class clazz, MetadataCacheConfig cacheConfig); + + /** + * Create a metadata cache specialized for a specific class. + * + * @param + * @param clazz + * the class type to be used for serialization/deserialization + * @return the metadata cache object + */ + default MetadataCache getMetadataCache(Class clazz) { + return getMetadataCache(clazz, getDefaultMetadataCacheConfig()); + } /** * Create a metadata cache specialized for a specific class. @@ -146,9 +160,23 @@ public interface MetadataStore extends AutoCloseable { * @param * @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 */ - MetadataCache getMetadataCache(TypeReference typeRef); + MetadataCache getMetadataCache(TypeReference typeRef, MetadataCacheConfig cacheConfig); + + /** + * Create a metadata cache specialized for a specific class. + * + * @param + * @param typeRef + * the type ref description to be used for serialization/deserialization + * @return the metadata cache object + */ + default MetadataCache getMetadataCache(TypeReference typeRef) { + return getMetadataCache(typeRef, getDefaultMetadataCacheConfig()); + } /** * Create a metadata cache that uses a particular serde object. @@ -156,7 +184,30 @@ public interface MetadataStore extends AutoCloseable { * @param * @param serde * the custom serialization/deserialization object + * @param cacheConfig + * the cache configuration to be used * @return the metadata cache object */ - MetadataCache getMetadataCache(MetadataSerde serde); + MetadataCache getMetadataCache(MetadataSerde serde, MetadataCacheConfig cacheConfig); + + /** + * Create a metadata cache that uses a particular serde object. + * + * @param + * @param serde + * the custom serialization/deserialization object + * @return the metadata cache object + */ + default MetadataCache getMetadataCache(MetadataSerde serde) { + return getMetadataCache(serde, getDefaultMetadataCacheConfig()); + } + + /** + * Returns the default metadata cache config. + * + * @return default metadata cache config + */ + default MetadataCacheConfig getDefaultMetadataCacheConfig() { + return MetadataCacheConfig.builder().build(); + } } diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/cache/impl/MetadataCacheImpl.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/cache/impl/MetadataCacheImpl.java index 67fa7afd00f4a..2cbe9a6dc19b3 100644 --- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/cache/impl/MetadataCacheImpl.java +++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/cache/impl/MetadataCacheImpl.java @@ -39,6 +39,7 @@ import org.apache.pulsar.metadata.api.CacheGetResult; import org.apache.pulsar.metadata.api.GetResult; import org.apache.pulsar.metadata.api.MetadataCache; +import org.apache.pulsar.metadata.api.MetadataCacheConfig; import org.apache.pulsar.metadata.api.MetadataSerde; import org.apache.pulsar.metadata.api.MetadataStore; import org.apache.pulsar.metadata.api.MetadataStoreException.AlreadyExistsException; @@ -50,30 +51,32 @@ @Slf4j public class MetadataCacheImpl implements MetadataCache, Consumer { - - private static final long CACHE_REFRESH_TIME_MILLIS = TimeUnit.MINUTES.toMillis(5); - @Getter private final MetadataStore store; private final MetadataSerde serde; private final AsyncLoadingCache>> objCache; - public MetadataCacheImpl(MetadataStore store, TypeReference typeRef) { - this(store, new JSONMetadataSerdeTypeRef<>(typeRef)); + public MetadataCacheImpl(MetadataStore store, TypeReference typeRef, MetadataCacheConfig cacheConfig) { + this(store, new JSONMetadataSerdeTypeRef<>(typeRef), cacheConfig); } - public MetadataCacheImpl(MetadataStore store, JavaType type) { - this(store, new JSONMetadataSerdeSimpleType<>(type)); + public MetadataCacheImpl(MetadataStore store, JavaType type, MetadataCacheConfig cacheConfig) { + this(store, new JSONMetadataSerdeSimpleType<>(type), cacheConfig); } - public MetadataCacheImpl(MetadataStore store, MetadataSerde serde) { + public MetadataCacheImpl(MetadataStore store, MetadataSerde serde, MetadataCacheConfig cacheConfig) { this.store = store; this.serde = serde; - this.objCache = Caffeine.newBuilder() - .refreshAfterWrite(CACHE_REFRESH_TIME_MILLIS, TimeUnit.MILLISECONDS) - .expireAfterWrite(CACHE_REFRESH_TIME_MILLIS * 2, TimeUnit.MILLISECONDS) + Caffeine cacheBuilder = Caffeine.newBuilder(); + if (cacheConfig.getRefreshAfterWriteMillis() > 0) { + cacheBuilder.refreshAfterWrite(cacheConfig.getRefreshAfterWriteMillis(), TimeUnit.MILLISECONDS); + } + if (cacheConfig.getExpireAfterWriteMillis() > 0) { + cacheBuilder.expireAfterWrite(cacheConfig.getExpireAfterWriteMillis(), TimeUnit.MILLISECONDS); + } + this.objCache = cacheBuilder .buildAsync(new AsyncCacheLoader>>() { @Override public CompletableFuture>> asyncLoad(String key, Executor executor) { diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/LeaderElectionImpl.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/LeaderElectionImpl.java index af873201e96c9..d903d8e2c051f 100644 --- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/LeaderElectionImpl.java +++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/LeaderElectionImpl.java @@ -32,6 +32,7 @@ import org.apache.bookkeeper.common.util.SafeRunnable; import org.apache.pulsar.metadata.api.GetResult; import org.apache.pulsar.metadata.api.MetadataCache; +import org.apache.pulsar.metadata.api.MetadataCacheConfig; import org.apache.pulsar.metadata.api.MetadataSerde; import org.apache.pulsar.metadata.api.MetadataStoreException; import org.apache.pulsar.metadata.api.MetadataStoreException.AlreadyClosedException; @@ -73,7 +74,9 @@ private enum InternalState { this.path = path; this.serde = new JSONMetadataSerdeSimpleType<>(TypeFactory.defaultInstance().constructSimpleType(clazz, null)); this.store = store; - this.cache = store.getMetadataCache(clazz); + this.cache = store.getMetadataCache(clazz, MetadataCacheConfig.builder() + .expireAfterWriteMillis(-1L) + .build()); this.leaderElectionState = LeaderElectionState.NoLeader; this.internalState = InternalState.Init; this.stateChangesListener = stateChangesListener; @@ -102,10 +105,13 @@ private synchronized CompletableFuture elect() { } else { return tryToBecomeLeader(); } - }).thenCompose(leaderElectionState -> - // make sure that the cache contains the current leader - // so that getLeaderValueIfPresent works on all brokers - cache.get(path).thenApply(__ -> leaderElectionState)); + }).thenComposeAsync(leaderElectionState -> { + // make sure that the cache contains the current leader + // so that getLeaderValueIfPresent works on all brokers + cache.refresh(path); + return cache.get(path) + .thenApply(__ -> leaderElectionState); + }, executor); } private synchronized CompletableFuture handleExistingLeaderValue(GetResult res) { @@ -207,11 +213,6 @@ private synchronized CompletableFuture tryToBecomeLeader() // 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 -> { diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/AbstractMetadataStore.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/AbstractMetadataStore.java index 8986811ad9f71..4b9ff914fcf74 100644 --- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/AbstractMetadataStore.java +++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/AbstractMetadataStore.java @@ -47,6 +47,7 @@ import org.apache.pulsar.common.util.FutureUtil; import org.apache.pulsar.metadata.api.GetResult; import org.apache.pulsar.metadata.api.MetadataCache; +import org.apache.pulsar.metadata.api.MetadataCacheConfig; import org.apache.pulsar.metadata.api.MetadataEvent; import org.apache.pulsar.metadata.api.MetadataEventSynchronizer; import org.apache.pulsar.metadata.api.MetadataSerde; @@ -212,23 +213,23 @@ protected boolean shouldIgnoreEvent(MetadataEvent event, GetResult existingValue } @Override - public MetadataCache getMetadataCache(Class clazz) { + public MetadataCache getMetadataCache(Class clazz, MetadataCacheConfig cacheConfig) { MetadataCacheImpl metadataCache = new MetadataCacheImpl(this, - TypeFactory.defaultInstance().constructSimpleType(clazz, null)); + TypeFactory.defaultInstance().constructSimpleType(clazz, null), cacheConfig); metadataCaches.add(metadataCache); return metadataCache; } @Override - public MetadataCache getMetadataCache(TypeReference typeRef) { - MetadataCacheImpl metadataCache = new MetadataCacheImpl(this, typeRef); + public MetadataCache getMetadataCache(TypeReference typeRef, MetadataCacheConfig cacheConfig) { + MetadataCacheImpl metadataCache = new MetadataCacheImpl(this, typeRef, cacheConfig); metadataCaches.add(metadataCache); return metadataCache; } @Override - public MetadataCache getMetadataCache(MetadataSerde serde) { - MetadataCacheImpl metadataCache = new MetadataCacheImpl<>(this, serde); + public MetadataCache getMetadataCache(MetadataSerde serde, MetadataCacheConfig cacheConfig) { + MetadataCacheImpl metadataCache = new MetadataCacheImpl<>(this, serde, cacheConfig); metadataCaches.add(metadataCache); return metadataCache; } diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/FaultInjectionMetadataStore.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/FaultInjectionMetadataStore.java index 679b49a740b8f..4972570654c1f 100644 --- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/FaultInjectionMetadataStore.java +++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/FaultInjectionMetadataStore.java @@ -31,6 +31,7 @@ import org.apache.pulsar.common.util.FutureUtil; import org.apache.pulsar.metadata.api.GetResult; import org.apache.pulsar.metadata.api.MetadataCache; +import org.apache.pulsar.metadata.api.MetadataCacheConfig; import org.apache.pulsar.metadata.api.MetadataSerde; import org.apache.pulsar.metadata.api.MetadataStoreException; import org.apache.pulsar.metadata.api.Notification; @@ -146,18 +147,18 @@ public void registerListener(Consumer listener) { } @Override - public MetadataCache getMetadataCache(Class clazz) { - return store.getMetadataCache(clazz); + public MetadataCache getMetadataCache(Class clazz, MetadataCacheConfig cacheConfig) { + return store.getMetadataCache(clazz, cacheConfig); } @Override - public MetadataCache getMetadataCache(TypeReference typeRef) { - return store.getMetadataCache(typeRef); + public MetadataCache getMetadataCache(TypeReference typeRef, MetadataCacheConfig cacheConfig) { + return store.getMetadataCache(typeRef, cacheConfig); } @Override - public MetadataCache getMetadataCache(MetadataSerde serde) { - return store.getMetadataCache(serde); + public MetadataCache getMetadataCache(MetadataSerde serde, MetadataCacheConfig cacheConfig) { + return store.getMetadataCache(serde, cacheConfig); } @Override From 02df95be14076573d3575463f9e00405815d079c Mon Sep 17 00:00:00 2001 From: Lari Hotari Date: Fri, 2 Sep 2022 15:29:38 +0300 Subject: [PATCH 2/3] Add test case --- .../broker/MultiBrokerTestZKBaseTest.java | 61 ++++++++++ ...ltiBrokerLeaderElectionExpirationTest.java | 114 ++++++++++++++++++ .../MultiBrokerLeaderElectionTest.java | 38 +----- 3 files changed, 177 insertions(+), 36 deletions(-) create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/MultiBrokerTestZKBaseTest.java create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/MultiBrokerLeaderElectionExpirationTest.java diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/MultiBrokerTestZKBaseTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/MultiBrokerTestZKBaseTest.java new file mode 100644 index 0000000000000..e6cf86c05b5eb --- /dev/null +++ b/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()); + } +} diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/MultiBrokerLeaderElectionExpirationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/MultiBrokerLeaderElectionExpirationTest.java new file mode 100644 index 0000000000000..65962868b87c9 --- /dev/null +++ b/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 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 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(); + } + } + } +} diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/MultiBrokerLeaderElectionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/MultiBrokerLeaderElectionTest.java index 462b640c17511..36ed1ca6c6190 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/MultiBrokerLeaderElectionTest.java +++ b/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; From ccd732a41c4459f2143e91200051b5fbb99ae137 Mon Sep 17 00:00:00 2001 From: Lari Hotari Date: Mon, 5 Sep 2022 16:27:41 +0300 Subject: [PATCH 3/3] Prevent stale values in leader election cache --- .../coordination/impl/LeaderElectionImpl.java | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/LeaderElectionImpl.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/LeaderElectionImpl.java index d903d8e2c051f..86fac33f7c2c9 100644 --- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/LeaderElectionImpl.java +++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/LeaderElectionImpl.java @@ -25,6 +25,7 @@ import java.util.concurrent.CompletionException; import java.util.concurrent.ExecutionException; import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import java.util.function.Consumer; import lombok.extern.slf4j.Slf4j; @@ -53,6 +54,7 @@ class LeaderElectionImpl implements LeaderElection { private final MetadataStoreExtended store; private final MetadataCache cache; private final Consumer stateChangesListener; + private final ScheduledFuture updateCachedValueFuture; private LeaderElectionState leaderElectionState; private Optional version = Optional.empty(); @@ -74,9 +76,10 @@ private enum InternalState { this.path = path; this.serde = new JSONMetadataSerdeSimpleType<>(TypeFactory.defaultInstance().constructSimpleType(clazz, null)); this.store = store; - this.cache = store.getMetadataCache(clazz, MetadataCacheConfig.builder() + MetadataCacheConfig metadataCacheConfig = MetadataCacheConfig.builder() .expireAfterWriteMillis(-1L) - .build()); + .build(); + this.cache = store.getMetadataCache(clazz, metadataCacheConfig); this.leaderElectionState = LeaderElectionState.NoLeader; this.internalState = InternalState.Init; this.stateChangesListener = stateChangesListener; @@ -84,6 +87,9 @@ private enum InternalState { store.registerListener(this::handlePathNotification); store.registerSessionListener(this::handleSessionNotification); + updateCachedValueFuture = executor.scheduleWithFixedDelay(SafeRunnable.safeRun(this::getLeaderValue), + metadataCacheConfig.getRefreshAfterWriteMillis() / 2, + metadataCacheConfig.getRefreshAfterWriteMillis(), TimeUnit.MILLISECONDS); } @Override @@ -230,6 +236,7 @@ private synchronized CompletableFuture tryToBecomeLeader() @Override public void close() throws Exception { + updateCachedValueFuture.cancel(true); try { asyncClose().join(); } catch (CompletionException e) {