diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java index 44048b78cf305..7a541d40880cf 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java @@ -38,6 +38,7 @@ import java.util.Objects; import java.util.Optional; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; @@ -74,7 +75,6 @@ import org.apache.pulsar.broker.intercept.BrokerInterceptor; import org.apache.pulsar.broker.intercept.BrokerInterceptors; import org.apache.pulsar.broker.loadbalance.LeaderElectionService; -import org.apache.pulsar.broker.loadbalance.LeaderElectionService.LeaderListener; import org.apache.pulsar.broker.loadbalance.LoadManager; import org.apache.pulsar.broker.loadbalance.LoadReportUpdaterTask; import org.apache.pulsar.broker.loadbalance.LoadResourceQuotaUpdaterTask; @@ -115,6 +115,12 @@ import org.apache.pulsar.functions.worker.ErrorNotifier; import org.apache.pulsar.functions.worker.WorkerConfig; import org.apache.pulsar.functions.worker.WorkerService; +import org.apache.pulsar.metadata.api.MetadataStoreConfig; +import org.apache.pulsar.metadata.api.MetadataStoreException; +import org.apache.pulsar.metadata.api.coordination.CoordinationService; +import org.apache.pulsar.metadata.api.coordination.LeaderElectionState; +import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended; +import org.apache.pulsar.metadata.coordination.impl.CoordinationServiceImpl; import org.apache.pulsar.packages.management.core.PackagesManagement; import org.apache.pulsar.packages.management.core.PackagesStorage; import org.apache.pulsar.packages.management.core.PackagesStorageProvider; @@ -209,6 +215,9 @@ public class PulsarService implements AutoCloseable { private PrometheusMetricsServlet metricsServlet; private List pendingMetricsProviders; + private MetadataStoreExtended localMetadataStore; + private CoordinationService coordinationService; + public enum State { Init, Started, Closed @@ -267,7 +276,7 @@ public PulsarService(ServiceConfiguration config, new DefaultThreadFactory("pulsar")); this.cacheExecutor = Executors.newScheduledThreadPool(config.getNumCacheExecutorThreadPoolSize(), new DefaultThreadFactory("zk-cache-callback")); - } + } /** * Close the current pulsar service. All resources are released. @@ -307,7 +316,7 @@ public void close() throws PulsarServerException { } if (this.leaderElectionService != null) { - this.leaderElectionService.stop(); + this.leaderElectionService.close(); this.leaderElectionService = null; } @@ -378,10 +387,25 @@ public void close() throws PulsarServerException { transactionBufferClient.close(); } + if (coordinationService != null) { + coordinationService.close(); + } + + if (localMetadataStore != null) { + localMetadataStore.close(); + } + state = State.Closed; isClosedCondition.signalAll(); } catch (Exception e) { - throw new PulsarServerException(e); + if (e instanceof CompletionException && e.getCause() instanceof MetadataStoreException) { + throw new PulsarServerException(MetadataStoreException.unwrap((CompletionException) e)); + } else if (e.getCause() instanceof CompletionException + && e.getCause().getCause() instanceof MetadataStoreException) { + throw new PulsarServerException(MetadataStoreException.unwrap((CompletionException) e.getCause())); + } else { + throw new PulsarServerException(e); + } } finally { mutex.unlock(); } @@ -440,6 +464,10 @@ public void start() throws PulsarServerException { throw new IllegalArgumentException("brokerServicePort/brokerServicePortTls must be present"); } + localMetadataStore = createLocalMetadataStore(); + + coordinationService = new CoordinationServiceImpl(localMetadataStore); + orderedExecutor = OrderedExecutor.newBuilder() .numThreads(config.getNumOrderedExecutorThreads()) .name("pulsar-ordered") @@ -646,36 +674,51 @@ public Boolean get() { } } - protected void startLeaderElectionService() { - this.leaderElectionService = new LeaderElectionService(this, new LeaderListener() { - @Override - public synchronized void brokerIsTheLeaderNow() { - if (getConfiguration().isLoadBalancerEnabled()) { - long loadSheddingInterval = TimeUnit.MINUTES - .toMillis(getConfiguration().getLoadBalancerSheddingIntervalMinutes()); - long resourceQuotaUpdateInterval = TimeUnit.MINUTES - .toMillis(getConfiguration().getLoadBalancerResourceQuotaUpdateIntervalMinutes()); - - loadSheddingTask = loadManagerExecutor.scheduleAtFixedRate(new LoadSheddingTask(loadManager), - loadSheddingInterval, loadSheddingInterval, TimeUnit.MILLISECONDS); - loadResourceQuotaTask = loadManagerExecutor.scheduleAtFixedRate( - new LoadResourceQuotaUpdaterTask(loadManager), resourceQuotaUpdateInterval, - resourceQuotaUpdateInterval, TimeUnit.MILLISECONDS); - } - } + public MetadataStoreExtended createLocalMetadataStore() throws MetadataStoreException { + return MetadataStoreExtended.create(config.getZookeeperServers(), + MetadataStoreConfig.builder() + .sessionTimeoutMillis((int) config.getZooKeeperSessionTimeoutMillis()) + .allowReadOnlyOperations(false) + .build()); + } - @Override - public synchronized void brokerIsAFollowerNow() { - if (loadSheddingTask != null) { - loadSheddingTask.cancel(false); - loadSheddingTask = null; - } - if (loadResourceQuotaTask != null) { - loadResourceQuotaTask.cancel(false); - loadResourceQuotaTask = null; - } - } - }); + protected void startLeaderElectionService() { + this.leaderElectionService = new LeaderElectionService(coordinationService, getSafeWebServiceAddress(), + state -> { + if (state == LeaderElectionState.Leading) { + LOG.info("This broker was elected leader"); + if (getConfiguration().isLoadBalancerEnabled()) { + long loadSheddingInterval = TimeUnit.MINUTES + .toMillis(getConfiguration().getLoadBalancerSheddingIntervalMinutes()); + long resourceQuotaUpdateInterval = TimeUnit.MINUTES + .toMillis(getConfiguration().getLoadBalancerResourceQuotaUpdateIntervalMinutes()); + + if (loadSheddingTask != null) { + loadSheddingTask.cancel(false); + } + if (loadResourceQuotaTask != null) { + loadResourceQuotaTask.cancel(false); + } + loadSheddingTask = loadManagerExecutor.scheduleAtFixedRate( + new LoadSheddingTask(loadManager), + loadSheddingInterval, loadSheddingInterval, TimeUnit.MILLISECONDS); + loadResourceQuotaTask = loadManagerExecutor.scheduleAtFixedRate( + new LoadResourceQuotaUpdaterTask(loadManager), resourceQuotaUpdateInterval, + resourceQuotaUpdateInterval, TimeUnit.MILLISECONDS); + } + } else { + LOG.info("This broker is a follower. Current leader is {}", + leaderElectionService.getCurrentLeader()); + if (loadSheddingTask != null) { + loadSheddingTask.cancel(false); + loadSheddingTask = null; + } + if (loadResourceQuotaTask != null) { + loadResourceQuotaTask.cancel(false); + loadResourceQuotaTask = null; + } + } + }); leaderElectionService.start(); } @@ -1313,6 +1356,14 @@ public Optional getBrokerListenPortTls() { return brokerService.getListenPortTls(); } + public MetadataStoreExtended getLocalMetadataStore() { + return localMetadataStore; + } + + public CoordinationService getCoordinationService() { + return coordinationService; + } + public static WorkerConfig initializeWorkerConfigFromBrokerConfig(ServiceConfiguration brokerConfig, String workerConfigFile) throws IOException { WorkerConfig workerConfig = WorkerConfig.load(workerConfigFile); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LeaderBroker.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LeaderBroker.java index 29c21b34199ea..7b31f86767c3c 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LeaderBroker.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LeaderBroker.java @@ -18,52 +18,17 @@ */ package org.apache.pulsar.broker.loadbalance; -import com.google.common.base.Objects; -import java.util.concurrent.atomic.AtomicBoolean; +import lombok.AllArgsConstructor; +import lombok.Data; +import lombok.NoArgsConstructor; /** * A class to hold the contents of the leader election node. Facilitates serialization and deserialization of the * information that might be added for leader broker in the future. - * - * */ +@Data +@AllArgsConstructor +@NoArgsConstructor public class LeaderBroker { - public final String serviceUrl; - - private AtomicBoolean isLeaderReady = new AtomicBoolean(false); - - // Need this default constructor for json conversion. Please do not remove this. - public LeaderBroker() { - this(null); - } - - public LeaderBroker(String serviceUrl) { - this.serviceUrl = serviceUrl; - } - - public String getServiceUrl() { - return this.serviceUrl; - } - - @Override - public int hashCode() { - return Objects.hashCode(serviceUrl); - } - - @Override - public boolean equals(Object obj) { - if (obj instanceof LeaderBroker) { - LeaderBroker other = (LeaderBroker) obj; - return Objects.equal(serviceUrl, other.serviceUrl); - } - return false; - } - - public boolean isLeaderReady() { - return isLeaderReady.get(); - } - - public void setLeaderReady(boolean isLeaderReady) { - this.isLeaderReady.compareAndSet(!isLeaderReady, isLeaderReady); - } + private String serviceUrl; } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LeaderElectionService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LeaderElectionService.java index 7e2e90cfd0c62..5f1fc9b5bb550 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LeaderElectionService.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LeaderElectionService.java @@ -18,191 +18,46 @@ */ package org.apache.pulsar.broker.loadbalance; -import static com.google.common.base.Preconditions.checkState; -import com.fasterxml.jackson.databind.ObjectMapper; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.atomic.AtomicBoolean; -import java.util.concurrent.atomic.AtomicReference; -import org.apache.bookkeeper.util.ZkUtils; -import org.apache.pulsar.broker.PulsarService; -import org.apache.zookeeper.CreateMode; -import org.apache.zookeeper.KeeperException.NoNodeException; -import org.apache.zookeeper.KeeperException.NodeExistsException; -import org.apache.zookeeper.WatchedEvent; -import org.apache.zookeeper.Watcher; -import org.apache.zookeeper.ZooDefs.Ids; -import org.apache.zookeeper.ZooKeeper; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; +import java.util.Optional; +import java.util.concurrent.CompletableFuture; +import java.util.function.Consumer; +import org.apache.pulsar.metadata.api.coordination.CoordinationService; +import org.apache.pulsar.metadata.api.coordination.LeaderElection; +import org.apache.pulsar.metadata.api.coordination.LeaderElectionState; /** * A class that provides way to elect the leader among brokers. - * - * */ -public class LeaderElectionService { +public class LeaderElectionService implements AutoCloseable { - private static final Logger log = LoggerFactory.getLogger(LeaderElectionService.class); private static final String ELECTION_ROOT = "/loadbalance/leader"; - private final PulsarService pulsar; - private final ExecutorService executor; - - private boolean stopped = true; - - private boolean elected = false; - - private final ZooKeeper zkClient; - - private final AtomicReference currentLeader = new AtomicReference(); - private final AtomicBoolean isLeader = new AtomicBoolean(); - - private final ObjectMapper jsonMapper; - - /** - * Interface which should be implemented by classes which are interested in the leader election. The listener gets - * called when current broker becomes the leader. - */ - public interface LeaderListener { - void brokerIsTheLeaderNow(); - - void brokerIsAFollowerNow(); - } - - private final LeaderListener leaderListener; - - public LeaderElectionService(PulsarService pulsar, LeaderListener leaderListener) { - this.pulsar = pulsar; - this.zkClient = pulsar.getZkClient(); - this.executor = pulsar.getExecutor(); - this.leaderListener = leaderListener; - this.jsonMapper = new ObjectMapper(); - } - - /** - * We try to get the data in the ELECTION_ROOT node. If the node is present (i.e. leader is present), we store it in - * the currentLeader and keep a watch on the election node. If we lose the leader, then watch gets triggered and we - * do the election again. If the node does not exist while getting the data, we get NoNodeException. This means, - * there is no leader and we create the node at ELECTION_ROOT and write the leader broker's service URL in the node. - * Once the leader is known, we call the listener method so that leader can take further actions. - */ - private void elect() { - try { - byte[] data = zkClient.getData(ELECTION_ROOT, new Watcher() { - @Override - public void process(WatchedEvent event) { - log.warn("Type of the event is [{}] and path is [{}]", event.getType(), event.getPath()); - switch (event.getType()) { - case NodeDeleted: - log.warn("Election node {} is deleted, attempting re-election...", event.getPath()); - if (event.getPath().equals(ELECTION_ROOT)) { - log.info("This should call elect again..."); - executor.execute(() -> { - // If the node is deleted, attempt the re-election - log.info("Broker [{}] is calling re-election from the thread", - pulsar.getSafeWebServiceAddress()); - elect(); - }); - } - break; - - default: - log.warn("Got something wrong on watch: {}", event); - break; - } - } - }, null); - - LeaderBroker leaderBroker = jsonMapper.readValue(data, LeaderBroker.class); - currentLeader.set(leaderBroker); - isLeader.set(false); - elected = true; - leaderListener.brokerIsAFollowerNow(); - - // If broker comes here it is a follower. Do nothing, wait for the watch to trigger - log.info("Broker [{}] is the follower now. Waiting for the watch to trigger...", - pulsar.getSafeWebServiceAddress()); - - } catch (NoNodeException nne) { - // There's no leader yet... try to become the leader - try { - // Create the root node and add current broker's URL as its contents - LeaderBroker leaderBroker = new LeaderBroker(pulsar.getSafeWebServiceAddress()); - ZkUtils.createFullPathOptimistic(pulsar.getLocalZkCache().getZooKeeper(), ELECTION_ROOT, - jsonMapper.writeValueAsBytes(leaderBroker), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL); + private final LeaderElection leaderElection; + private final LeaderBroker localValue; - // Update the current leader and set the flag to true - currentLeader.set(new LeaderBroker(leaderBroker.getServiceUrl())); - isLeader.set(true); - elected = true; - - // Notify the listener that this broker is now the leader so that it can collect usage and start load - // manager. - log.info("Broker [{}] is the leader now, notifying the listener...", pulsar.getSafeWebServiceAddress()); - leaderListener.brokerIsTheLeaderNow(); - } catch (NodeExistsException nee) { - // Re-elect the new leader - log.warn( - "Got exception [{}] while creating election node because it already exists." - + " Attempting re-election...", - nee.getMessage()); - executor.execute(this::elect); - } catch (Exception e) { - // Kill the broker because this broker's session with zookeeper might be stale. Killing the broker will - // make sure that we get the fresh zookeeper session. - log.error("Got exception [{}] while creating the election node", e.getMessage()); - pulsar.getShutdownService().shutdown(-1); - } - - } catch (Exception e) { - // Kill the broker - log.error("Could not get the content of [{}], got exception [{}]. Shutting down the broker...", - ELECTION_ROOT, e); - pulsar.getShutdownService().shutdown(-1); - } + public LeaderElectionService(CoordinationService cs, String localWebServiceAddress, + Consumer listener) { + this.leaderElection = cs.getLeaderElection(LeaderBroker.class, ELECTION_ROOT, listener); + this.localValue = new LeaderBroker(localWebServiceAddress); } public void start() { - checkState(stopped); - stopped = false; - log.info("LeaderElectionService started"); - elect(); + leaderElection.elect(localValue).join(); } - public void stop() { - if (stopped) { - return; - } - - if (isLeader()) { - // Make sure to remove the leader election z-node in case the session doesn't - // get closed properly. This is to avoid having to wait the session timeout - // to elect a new one. - // This delete operation is safe to do here (with version=-1) because either: - // 1. The ZK session is still valid, in which case this broker is still - // the "leader" and we have to remove the z-node - // 2. The session has already expired, in which case this delete operation - // will not go through - try { - pulsar.getLocalZkCache().getZooKeeper().delete(ELECTION_ROOT, -1); - } catch (Throwable t) { - log.warn("Failed to cleanup election root znode", t); - } - } - stopped = true; - log.info("LeaderElectionService stopped"); + public void close() throws Exception { + leaderElection.close(); } - public LeaderBroker getCurrentLeader() { - return currentLeader.get(); + public CompletableFuture> readCurrentLeader() { + return leaderElection.getLeaderValue(); } - public boolean isLeader() { - return isLeader.get(); + public Optional getCurrentLeader() { + return leaderElection.getLeaderValueIfPresent(); } - public boolean isElected() { - return elected; + public boolean isLeader() { + return leaderElection.getState() == LeaderElectionState.Leading; } - } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java index 96ad4fde5970c..a0be2acdf3323 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java @@ -58,6 +58,7 @@ import org.apache.pulsar.broker.PulsarService; import org.apache.pulsar.broker.ServiceConfiguration; import org.apache.pulsar.broker.admin.AdminResource; +import org.apache.pulsar.broker.loadbalance.LeaderBroker; import org.apache.pulsar.broker.loadbalance.LeaderElectionService; import org.apache.pulsar.broker.loadbalance.LoadManager; import org.apache.pulsar.broker.loadbalance.ResourceUnit; @@ -447,7 +448,7 @@ private CompletableFuture> findBrokerServiceUrl( private void searchForCandidateBroker(NamespaceBundle bundle, CompletableFuture> lookupFuture, LookupOptions options) { - if (null == pulsar.getLeaderElectionService() || !pulsar.getLeaderElectionService().isElected()) { + if (null == pulsar.getLeaderElectionService()) { LOG.warn("The leader election has not yet been completed! NamespaceBundle[{}]", bundle); lookupFuture.completeExceptionally( new IllegalStateException("The leader election has not yet been completed!")); @@ -478,14 +479,17 @@ private void searchForCandidateBroker(NamespaceBundle bundle, } if (candidateBroker == null) { + Optional currentLeader = pulsar.getLeaderElectionService().getCurrentLeader(); + if (options.isAuthoritative()) { // leader broker already assigned the current broker as owner candidateBroker = pulsar.getSafeWebServiceAddress(); } else if (!this.loadManager.get().isCentralized() || pulsar.getLeaderElectionService().isLeader() + || !currentLeader.isPresent() // If leader is not active, fallback to pick the least loaded from current broker loadmanager - || !isBrokerActive(pulsar.getLeaderElectionService().getCurrentLeader().getServiceUrl()) + || !isBrokerActive(currentLeader.get().getServiceUrl()) ) { Optional availableBroker = getLeastLoadedFromLoadManager(bundle); if (!availableBroker.isPresent()) { @@ -496,7 +500,7 @@ private void searchForCandidateBroker(NamespaceBundle bundle, authoritativeRedirect = true; } else { // forward to leader broker to make assignment - candidateBroker = pulsar.getLeaderElectionService().getCurrentLeader().getServiceUrl(); + candidateBroker = currentLeader.get().getServiceUrl(); } } } catch (Exception e) { diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java index 5b5fb5009002e..f10431ca8f113 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java @@ -795,12 +795,7 @@ protected boolean isLeaderBroker() { } protected static boolean isLeaderBroker(PulsarService pulsar) { - - String leaderAddress = pulsar.getLeaderElectionService().getCurrentLeader().getServiceUrl(); - - String myAddress = pulsar.getSafeWebServiceAddress(); - - return myAddress.equals(leaderAddress); // If i am the leader, my decisions are + return pulsar.getLeaderElectionService().isLeader(); } // Non-Usual HTTP error codes diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java index fa3ddb2cf957c..47c8fe3764088 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java @@ -57,6 +57,7 @@ import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.common.policies.data.ClusterData; import org.apache.pulsar.common.policies.data.TenantInfo; +import org.apache.pulsar.metadata.impl.ZKMetadataStore; import org.apache.pulsar.zookeeper.ZooKeeperClientFactory; import org.apache.pulsar.zookeeper.ZookeeperClientFactoryImpl; import org.apache.zookeeper.CreateMode; @@ -230,7 +231,7 @@ protected final void internalCleanup() throws Exception { } bkExecutor = null; } - + } protected abstract void setup() throws Exception; @@ -281,6 +282,7 @@ protected void setupBrokerMocks(PulsarService pulsar) throws Exception { // Override default providers with mocked ones doReturn(mockZooKeeperClientFactory).when(pulsar).getZooKeeperClientFactory(); doReturn(mockBookKeeperClientFactory).when(pulsar).newBookKeeperClientFactory(); + doReturn(new ZKMetadataStore(mockZooKeeper)).when(pulsar).createLocalMetadataStore(); Supplier namespaceServiceSupplier = () -> spy(new NamespaceService(pulsar)); doReturn(namespaceServiceSupplier).when(pulsar).getNamespaceServiceProvider(); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/LeaderElectionServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/LeaderElectionServiceTest.java index 11e0aafcf27dc..3d76db7756347 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/LeaderElectionServiceTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/LeaderElectionServiceTest.java @@ -19,6 +19,9 @@ package org.apache.pulsar.broker.loadbalance; import com.google.common.collect.Sets; +import java.util.Optional; +import java.util.concurrent.TimeUnit; +import java.util.function.Consumer; import lombok.extern.slf4j.Slf4j; import org.apache.pulsar.broker.PulsarServerException; import org.apache.pulsar.broker.PulsarService; @@ -32,41 +35,22 @@ import org.apache.pulsar.common.policies.data.TenantInfo; import org.apache.pulsar.functions.worker.WorkerService; import org.apache.pulsar.zookeeper.LocalBookkeeperEnsemble; -import org.apache.pulsar.zookeeper.ZooKeeperCache; import org.mockito.Mockito; import org.testng.Assert; import org.testng.annotations.AfterMethod; import org.testng.annotations.BeforeMethod; import org.testng.annotations.Test; -import java.util.Optional; -import java.util.concurrent.Executors; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.TimeUnit; -import java.util.function.Consumer; - @Slf4j public class LeaderElectionServiceTest { private LocalBookkeeperEnsemble bkEnsemble; - private LeaderElectionService.LeaderListener listener; @BeforeMethod public void setup() throws Exception { bkEnsemble = new LocalBookkeeperEnsemble(3, 0, () -> 0); bkEnsemble.start(); log.info("---- bk started ----"); - listener = new LeaderElectionService.LeaderListener() { - @Override - public void brokerIsTheLeaderNow() { - log.info("i am a leader"); - } - - @Override - public void brokerIsAFollowerNow() { - log.info("i am a follower"); - } - }; } @AfterMethod(alwaysRun = true) @@ -75,43 +59,6 @@ void shutdown() throws Exception { log.info("---- bk stopped ----"); } - - @Test - public void electedShouldBeTrue() { - final ScheduledExecutorService ses = Executors.newSingleThreadScheduledExecutor(); - final String safeWebServiceAddress = "http://localhost:8080"; - ZooKeeperCache zkCache = Mockito.mock(ZooKeeperCache.class); - PulsarService pulsar = Mockito.mock(PulsarService.class); - - Mockito.when(pulsar.getZkClient()).thenReturn(bkEnsemble.getZkClient()); - Mockito.when(pulsar.getExecutor()).thenReturn(ses); - Mockito.when(pulsar.getSafeWebServiceAddress()).thenReturn(safeWebServiceAddress); - - Mockito.when(zkCache.getZooKeeper()).thenReturn(bkEnsemble.getZkClient()); - Mockito.when(pulsar.getLocalZkCache()).thenReturn(zkCache); - - LeaderElectionService leaderElectionService = new LeaderElectionService(pulsar, listener); - leaderElectionService.start(); - Assert.assertTrue(leaderElectionService.isElected()); - Assert.assertTrue(leaderElectionService.isLeader()); - Assert.assertEquals(leaderElectionService.getCurrentLeader().getServiceUrl(), safeWebServiceAddress); - log.info("leader state {} {} {}", - leaderElectionService.isElected(), - leaderElectionService.isLeader(), - leaderElectionService.getCurrentLeader().getServiceUrl()); - - LeaderElectionService followerElectionService = new LeaderElectionService(pulsar, listener); - followerElectionService.start(); - Assert.assertTrue(followerElectionService.isElected()); - Assert.assertFalse(followerElectionService.isLeader()); - Assert.assertEquals(followerElectionService.getCurrentLeader().getServiceUrl(), safeWebServiceAddress); - log.info("follower state {} {} {}", - followerElectionService.isElected(), - followerElectionService.isLeader(), - followerElectionService.getCurrentLeader().getServiceUrl()); - ses.shutdown(); - } - @Test public void anErrorShouldBeThrowBeforeLeaderElected() throws PulsarServerException, PulsarClientException, PulsarAdminException { final String clusterName = "elect-test"; @@ -142,14 +89,12 @@ public void anErrorShouldBeThrowBeforeLeaderElected() throws PulsarServerExcepti // broker, webService and leaderElectionService is started, but elect not ready; LeaderElectionService leaderElectionService = Mockito.mock(LeaderElectionService.class); - Mockito.when(leaderElectionService.isElected()).thenReturn(false); Mockito.doReturn(leaderElectionService).when(pulsar).getLeaderElectionService(); checkLookupException(tenant, namespace, client); // broker, webService and leaderElectionService is started, and elect is done; - Mockito.when(leaderElectionService.isElected()).thenReturn(true); Mockito.when(leaderElectionService.isLeader()).thenReturn(true); - Mockito.when(leaderElectionService.getCurrentLeader()).thenReturn(new LeaderBroker("http://localhost:8080")); + Mockito.when(leaderElectionService.getCurrentLeader()).thenReturn(Optional.of(new LeaderBroker("http://localhost:8080"))); Producer producer = client.newProducer() .topic("persistent://" + tenant + "/" + namespace + "/1p") diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/LoadBalancerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/LoadBalancerTest.java index fefbd7137c153..5118abadcf08f 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/LoadBalancerTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/LoadBalancerTest.java @@ -182,7 +182,7 @@ private LeaderBroker loopUntilLeaderChanges(LeaderElectionService les, LeaderBro while (loopCount < MAX_RETRIES) { Thread.sleep(1000); // Check if the new leader is elected. If yes, break without incrementing the loopCount - newLeader = les.getCurrentLeader(); + newLeader = les.getCurrentLeader().get(); if (newLeader.equals(oldLeader) == false) { break; } @@ -720,7 +720,7 @@ public void testLeaderElection() throws Exception { activePulsar.add(allServices[j]); LeaderElectionService les = allServices[j].getLeaderElectionService(); if (les.isLeader()) { - oldLeader = les.getCurrentLeader(); + oldLeader = les.getCurrentLeader().get(); leaderPulsar = allServices[j]; // set the refence to null in the main array, // in order to prevent closing this PulsarService twice @@ -733,7 +733,8 @@ public void testLeaderElection() throws Exception { // Make sure both brokers see the same leader log.info("Old leader is : {}", oldLeader.getServiceUrl()); for (PulsarService pulsar : activePulsar) { - assertEquals(pulsar.getLeaderElectionService().getCurrentLeader(), oldLeader); + log.info("Current leader for {} is : {}", pulsar.getWebServiceAddress(), pulsar.getLeaderElectionService().getCurrentLeader()); + assertEquals(pulsar.getLeaderElectionService().readCurrentLeader().join(), Optional.of(oldLeader)); } // Do leader election by killing the leader broker diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/OwnerShipForCurrentServerTestBase.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/OwnerShipForCurrentServerTestBase.java index 98d427719c70a..1b0c7f3506cd3 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/OwnerShipForCurrentServerTestBase.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/OwnerShipForCurrentServerTestBase.java @@ -35,6 +35,7 @@ import org.apache.pulsar.broker.intercept.CounterBrokerInterceptor; import org.apache.pulsar.client.admin.PulsarAdmin; import org.apache.pulsar.client.api.PulsarClient; +import org.apache.pulsar.metadata.impl.ZKMetadataStore; import org.apache.pulsar.zookeeper.ZooKeeperClientFactory; import org.apache.pulsar.zookeeper.ZookeeperClientFactoryImpl; import org.apache.zookeeper.CreateMode; @@ -127,6 +128,7 @@ protected void setupBrokerMocks(PulsarService pulsar) throws Exception { // Override default providers with mocked ones doReturn(mockZooKeeperClientFactory).when(pulsar).getZooKeeperClientFactory(); doReturn(mockBookKeeperClientFactory).when(pulsar).newBookKeeperClientFactory(); + doReturn(new ZKMetadataStore(mockZooKeeper)).when(pulsar).createLocalMetadataStore(); Supplier namespaceServiceSupplier = () -> spy(new NamespaceService(pulsar)); doReturn(namespaceServiceSupplier).when(pulsar).getNamespaceServiceProvider(); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ConsumedLedgersTrimTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ConsumedLedgersTrimTest.java index 678d9a5657ca4..9cf2ee7bc539a 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ConsumedLedgersTrimTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ConsumedLedgersTrimTest.java @@ -31,11 +31,11 @@ import org.apache.pulsar.client.api.Message; import org.apache.pulsar.client.api.MessageId; import org.apache.pulsar.client.api.Producer; -import org.junit.Test; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.testng.Assert; import org.testng.annotations.AfterMethod; +import org.testng.annotations.Test; public class ConsumedLedgersTrimTest extends BrokerTestBase { diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTestBase.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTestBase.java index d061c827c354a..6a2cfaaeac1a3 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTestBase.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTestBase.java @@ -49,6 +49,7 @@ import org.apache.pulsar.broker.namespace.NamespaceService; import org.apache.pulsar.client.admin.PulsarAdmin; import org.apache.pulsar.client.api.PulsarClient; +import org.apache.pulsar.metadata.impl.ZKMetadataStore; import org.apache.pulsar.zookeeper.ZooKeeperClientFactory; import org.apache.pulsar.zookeeper.ZookeeperClientFactoryImpl; import org.apache.zookeeper.CreateMode; @@ -131,6 +132,7 @@ protected void setupBrokerMocks(PulsarService pulsar) throws Exception { // Override default providers with mocked ones doReturn(mockZooKeeperClientFactory).when(pulsar).getZooKeeperClientFactory(); doReturn(mockBookKeeperClientFactory).when(pulsar).newBookKeeperClientFactory(); + doReturn(new ZKMetadataStore(mockZooKeeper)).when(pulsar).createLocalMetadataStore(); Supplier namespaceServiceSupplier = () -> spy(new NamespaceService(pulsar)); doReturn(namespaceServiceSupplier).when(pulsar).getNamespaceServiceProvider(); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/WebServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/WebServiceTest.java index 6816fb0c55517..04f9d81107810 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/WebServiceTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/WebServiceTest.java @@ -65,8 +65,10 @@ import org.apache.pulsar.common.policies.data.TenantInfo; import org.apache.pulsar.common.util.ObjectMapperFactory; import org.apache.pulsar.common.util.SecurityUtility; +import org.apache.pulsar.metadata.impl.ZKMetadataStore; import org.apache.pulsar.zookeeper.MockedZooKeeperClientFactoryImpl; import org.apache.zookeeper.CreateMode; +import org.apache.zookeeper.MockZooKeeper; import org.apache.zookeeper.ZooDefs; import org.asynchttpclient.AsyncHttpClient; import org.asynchttpclient.BoundRequestBuilder; @@ -362,6 +364,7 @@ private void setupEnv(boolean enableFilter, String minApiVersion, boolean allowU pulsar = spy(new PulsarService(config)); doReturn(zkFactory).when(pulsar).getZooKeeperClientFactory(); + doReturn(new ZKMetadataStore(MockZooKeeper.newInstance())).when(pulsar).createLocalMetadataStore(); doReturn(new MockedBookKeeperClientFactory()).when(pulsar).newBookKeeperClientFactory(); pulsar.start(); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/AuthenticatedProducerConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/AuthenticatedProducerConsumerTest.java index b40cc6cb92e11..7f3e7b3e06e05 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/AuthenticatedProducerConsumerTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/AuthenticatedProducerConsumerTest.java @@ -40,6 +40,7 @@ import org.apache.pulsar.common.policies.data.AuthAction; import org.apache.pulsar.common.policies.data.ClusterData; import org.apache.pulsar.common.policies.data.TenantInfo; +import org.apache.zookeeper.KeeperException.Code; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.testng.Assert; @@ -315,7 +316,7 @@ public void testInternalServerExceptionOnLookup() throws Exception { String topic = "persistent://" + namespace + "1/topic1"; // this will cause NPE and it should throw 500 - mockZooKeeperGlobal.shutdown(); + mockZooKeeperGlobal.setAlwaysFail(Code.SESSIONEXPIRED); pulsar.getConfiguration().setSuperUserRoles(Sets.newHashSet()); try { admin.topics().getPartitionedTopicMetadata(topic); @@ -328,6 +329,7 @@ public void testInternalServerExceptionOnLookup() throws Exception { Assert.assertTrue(e.getCause() instanceof InternalServerErrorException); } + mockZooKeeperGlobal.unsetAlwaysFail(); } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/BrokerServiceLookupTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/BrokerServiceLookupTest.java index 03af0264d75f2..dfa1c42b26df6 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/BrokerServiceLookupTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/BrokerServiceLookupTest.java @@ -999,15 +999,15 @@ public void testModularLoadManagerSplitBundle() throws Exception { pulsar2.getLoadManager().get().writeLoadReportOnZookeeper(); // (5) Modular-load-manager will split the bundle due to max-topic threshold reached - Field leaderField = LeaderElectionService.class.getDeclaredField("isLeader"); Method updateAllMethod = ModularLoadManagerImpl.class.getDeclaredMethod("updateAll"); updateAllMethod.setAccessible(true); - leaderField.setAccessible(true); - AtomicBoolean isLeader = (AtomicBoolean) leaderField.get(pulsar2.getLeaderElectionService()); - isLeader.set(true); + + // broker-2 loadManager is a leader and let it refresh load-report from all the brokers + pulsar.getLeaderElectionService().close(); + ModularLoadManagerImpl loadManager = (ModularLoadManagerImpl) ((ModularLoadManagerWrapper) pulsar2 .getLoadManager().get()).getLoadManager(); - // broker-2 loadManager is a leader and let it refresh load-report from all the brokers + updateAllMethod.invoke(loadManager); conf2.setLoadBalancerAutoBundleSplitEnabled(true); conf2.setLoadBalancerAutoUnloadSplitBundlesEnabled(true);