diff --git a/bin/pulsar b/bin/pulsar index 884c44fdf6765..06bf2ea156c95 100755 --- a/bin/pulsar +++ b/bin/pulsar @@ -21,6 +21,7 @@ PULSAR_HOME=`cd $BINDIR/..;pwd` DEFAULT_BROKER_CONF=$PULSAR_HOME/conf/broker.conf DEFAULT_BOOKKEEPER_CONF=$PULSAR_HOME/conf/bookkeeper.conf DEFAULT_ZK_CONF=$PULSAR_HOME/conf/zookeeper.conf +DEFAULT_DATA_ZK_CONF=$PULSAR_HOME/conf/data_zookeeper.conf DEFAULT_GLOBAL_ZK_CONF=$PULSAR_HOME/conf/global_zookeeper.conf DEFAULT_DISCOVERY_CONF=$PULSAR_HOME/conf/discovery.conf DEFAULT_STANDALONE_CONF=$PULSAR_HOME/conf/standalone.conf @@ -83,6 +84,7 @@ Environment variables: PULSAR_BROKER_CONF Configuration file for broker (default: $DEFAULT_BROKER_CONF) PULSAR_BOOKKEEPER_CONF Configuration file for bookie (default: $DEFAULT_BOOKKEEPER_CONF) PULSAR_ZK_CONF Configuration file for zookeeper (default: $DEFAULT_ZK_CONF) + PULSAR_DATA_ZK_CONF Configuration file for data zookeeper (default: $DEFAULT_DATA_ZK_CONF) PULSAR_GLOBAL_ZK_CONF Configuration file for global zookeeper (default: $DEFAULT_GLOBAL_ZK_CONF) PULSAR_DISCOVERY_CONF Configuration file for discovery service (default: $DEFAULT_DISCOVERY_CONF) PULSAR_WEBSOCKET_CONF Configuration file for websocket proxy (default: $DEFAULT_WEBSOCKET_CONF) @@ -141,6 +143,10 @@ if [ -z "$PULSAR_ZK_CONF" ]; then PULSAR_ZK_CONF=$DEFAULT_ZK_CONF fi +if [ -z "$PULSAR_DATA_ZK_CONF" ]; then + PULSAR_DATA_ZK_CONF=$DEFAULT_DATA_ZK_CONF +fi + if [ -z "$PULSAR_GLOBAL_ZK_CONF" ]; then PULSAR_GLOBAL_ZK_CONF=$DEFAULT_GLOBAL_ZK_CONF fi @@ -193,6 +199,9 @@ elif [ $COMMAND == "bookie" ]; then elif [ $COMMAND == "zookeeper" ]; then PULSAR_LOG_FILE=${PULSAR_LOG_FILE:-"zookeeper.log"} exec $JAVA $OPTS -Dpulsar.log.file=$PULSAR_LOG_FILE org.apache.zookeeper.server.quorum.QuorumPeerMain $PULSAR_ZK_CONF $@ +elif [ $COMMAND == "data-zookeeper" ]; then + PULSAR_LOG_FILE=${PULSAR_LOG_FILE:-"data-zookeeper.log"} + exec $JAVA $OPTS -Dpulsar.log.file=$PULSAR_LOG_FILE org.apache.zookeeper.server.quorum.QuorumPeerMain $PULSAR_DATA_ZK_CONF $@ elif [ $COMMAND == "global-zookeeper" ]; then PULSAR_LOG_FILE=${PULSAR_LOG_FILE:-"global-zookeeper.log"} # Allow global ZK to turn into read-only mode when it cannot reach the quorum diff --git a/conf/broker.conf b/conf/broker.conf index 73a4402da8b86..5c801f6869ec1 100644 --- a/conf/broker.conf +++ b/conf/broker.conf @@ -19,6 +19,9 @@ # Zookeeper quorum connection string zookeeperServers= +# Data Zookeeper quorum connection string +dataZookeeperServers= + # Global Zookeeper quorum connection string globalZookeeperServers= @@ -49,6 +52,9 @@ clusterName= # Zookeeper session timeout in milliseconds zooKeeperSessionTimeoutMillis=30000 +# Data Zookeeper session timeout in milliseconds +dataZooKeeperSessionTimeoutMillis=60000 + # Time to wait for broker graceful shutdown. After this time elapses, the process will be killed brokerShutdownTimeoutMs=3000 @@ -252,6 +258,12 @@ loadBalancerNamespaceBundleMaxBandwidthMbytes=100 # maximum number of bundles in a namespace loadBalancerNamespaceMaximumBundles=128 +# OwnershipCacheFactory-provider class name if requires to inject non-default namespace-bundle ownership implementation +namespaceOwnershipCacheFactoryProvider= + +# LoadBalancerFactory-provider class name if requires to inject non-default LoadBalancer implementation +loadBalancerFactoryProvider= + ### --- Replication --- ### # Enable replication metrics diff --git a/conf/data_zookeeper.conf b/conf/data_zookeeper.conf new file mode 100644 index 0000000000000..3477c2c3c642c --- /dev/null +++ b/conf/data_zookeeper.conf @@ -0,0 +1,42 @@ +# +# Copyright 2016 Yahoo Inc. +# +# Licensed 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. +# + +# The number of milliseconds of each tick +tickTime=2000 +# The number of ticks that the initial +# synchronization phase can take +initLimit=10 +# The number of ticks that can pass between +# sending a request and getting an acknowledgement +syncLimit=5 +# the directory where the snapshot is stored. +dataDir=data/zookeeper +# the port at which the clients will connect +clientPort=2181 +# the maximum number of client connections. +# increase this if you need to handle more clients +#maxClientCnxns=60 +# +# Be sure to read the maintenance section of the +# administrator guide before turning on autopurge. +# +# http://zookeeper.apache.org/doc/current/zookeeperAdmin.html#sc_maintenance +# +# The number of snapshots to retain in dataDir +autopurge.snapRetainCount=3 +# Purge task interval in hours +# Set to "0" to disable auto purge feature +autopurge.purgeInterval=1 diff --git a/conf/standalone.conf b/conf/standalone.conf index c7efc28cbab96..a3c9bce488503 100644 --- a/conf/standalone.conf +++ b/conf/standalone.conf @@ -19,6 +19,9 @@ # Zookeeper quorum connection string zookeeperServers= +# Data Zookeeper quorum connection string +dataZookeeperServers= + # Global Zookeeper quorum connection string globalZookeeperServers= @@ -39,6 +42,9 @@ clusterName=standalone # Zookeeper session timeout in milliseconds zooKeeperSessionTimeoutMillis=30000 +# Data Zookeeper session timeout in milliseconds +dataZooKeeperSessionTimeoutMillis=60000 + # Time to wait for broker graceful shutdown. After this time elapses, the process will be killed brokerShutdownTimeoutMs=3000 @@ -225,6 +231,12 @@ loadBalancerNamespaceBundleMaxBandwidthMbytes=100 # maximum number of bundles in a namespace loadBalancerNamespaceMaximumBundles=128 +# OwnershipCacheFactory-provider class name if requires to inject non-default namespace-bundle ownership implementation +namespaceOwnershipCacheFactoryProvider= + +# LoadBalancerFactory-provider class name if requires to inject non-default LoadBalancer implementation +loadBalancerFactoryProvider= + ### --- Replication --- ### # Enable replication metrics diff --git a/docs/Architecture.md b/docs/Architecture.md index 02af179130738..e981376f3fb3c 100644 --- a/docs/Architecture.md +++ b/docs/Architecture.md @@ -71,6 +71,7 @@ Internally, a single managed ledger uses multiple Bookkeeper ledgers to store th Pulsar uses Apache Zookeeper for metadata, cluster configuration and coordination. - *Global Zookeeper* stores user provisioning data like properties, namespaces and policies which should be global consistent. - Each cluster has a *local zookeeper* ensemble which stores cluster specific configuration and coordination data, like ownership metadata, broker load reports, bookkeeper ledgers' metadata. +- *local zookeeper* can be highly available by configuring separete *data zookeeper* which stores bookkeeper ledgers' metadata and *local zookeeper* needs to store only cluster management configuration. ## Design diff --git a/docs/ClusterSetup.md b/docs/ClusterSetup.md index 122b3cff39eb8..1af40fbac8f42 100644 --- a/docs/ClusterSetup.md +++ b/docs/ClusterSetup.md @@ -46,6 +46,32 @@ Start ZK service on all the hosts: ```shell $ bin/pulsar-daemon start zookeeper ``` +#### Data ZooKeeper + +Start dedicated zookeeper quorum to store bookkeeper ledgers' metadata. + +##### Single zookeeper quorum +This does not require to configure and start _data_ zookeeper quorum, and pulsar instance uses _local_ zk quorum to store both cluster-management configuration and bookkeeper ledgers' metadata. + +##### Dedicated data-zookeeper quorum +In this scenario dedicated _data_ zk stores bookkeeper ledgers' metadata and _local_ zk stores only cluster-management configuration. Using dedicated _data_ zk makes _local_ zk and broker highly available because broker's availability only depends on cluster-management configuration that is stored into _local_ zk. + +Add all ZK servers the quorum configuration. Edit `conf/data_zookeeper.conf` and add +the following lines in all the ZK servers: + +``` +server.1=zk4.us-west.example.com:2888:3888 +server.2=zk5.us-west.example.com:2888:3888 +server.3=zk6.us-west.example.com:2888:3888 +... +``` + +Start ZK service on all the hosts: + +```shell +$ bin/pulsar-daemon start data-zookeeper +``` + #### Global ZooKeeper @@ -135,6 +161,7 @@ as well as the Pulsar metadata. ```shell $ bin/pulsar initialize-cluster-metadata --cluster us-west \ --zookeeper zk1.us-west.example.com:2181 \ + --data-zookeeper zk1.us-west.example.com:2181 \ --global-zookeeper zk1.us-west.example.com:2184 \ --service-url http://pulsar.us-west.example.com:8080/ \ --service-url-tls https://pulsar.us-west.example.com:8443/ diff --git a/pulsar-broker-common/src/main/java/com/yahoo/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/com/yahoo/pulsar/broker/ServiceConfiguration.java index ab3099362e5d5..0f1206690c6d5 100644 --- a/pulsar-broker-common/src/main/java/com/yahoo/pulsar/broker/ServiceConfiguration.java +++ b/pulsar-broker-common/src/main/java/com/yahoo/pulsar/broker/ServiceConfiguration.java @@ -36,6 +36,9 @@ public class ServiceConfiguration implements PulsarConfiguration{ // Zookeeper quorum connection string @FieldContext(required = true) private String zookeeperServers; + // Data Zookeeper quorum connection string + @FieldContext(required = false) + private String dataZookeeperServers; // Global Zookeeper quorum connection string @FieldContext(required = false) private String globalZookeeperServers; @@ -60,6 +63,8 @@ public class ServiceConfiguration implements PulsarConfiguration{ private String clusterName; // Zookeeper session timeout in milliseconds private long zooKeeperSessionTimeoutMillis = 30000; + // Data Zookeeper session timeout in milliseconds + private long dataZooKeeperSessionTimeoutMillis = 60000; // Time to wait for broker graceful shutdown. After this time elapses, the // process will be killed private long brokerShutdownTimeoutMs = 3000; @@ -203,7 +208,7 @@ public class ServiceConfiguration implements PulsarConfiguration{ private int loadBalancerBrokerOverloadedThresholdPercentage = 85; // interval to flush dynamic resource quota to ZooKeeper private int loadBalancerResourceQuotaUpdateIntervalMinutes = 15; - // Usage threshold to defermine a broker is having just right level of load + // Usage threshold to determine a broker is having just right level of load private int loadBalancerBrokerComfortLoadLevelPercentage = 65; // enable/disable automatic namespace bundle split private boolean loadBalancerAutoBundleSplitEnabled = false; @@ -218,6 +223,13 @@ public class ServiceConfiguration implements PulsarConfiguration{ // maximum number of bundles in a namespace private int loadBalancerNamespaceMaximumBundles = 128; + @FieldContext(required = false) + // OwnershipCacheFactory-provider class name if requires to inject non-default namespace-bundle ownership implementation + private String namespaceOwnershipCacheFactoryProvider; + @FieldContext(required = false) + // LoadBalancerFactory-provider class name if requires to inject non-default LoadBalancer implementation + private String loadBalancerFactoryProvider; + /**** --- Replication --- ****/ // Enable replication metrics private boolean replicationMetricsEnabled = false; @@ -249,6 +261,19 @@ public void setZookeeperServers(String zookeeperServers) { this.zookeeperServers = zookeeperServers; } + public String getDataZookeeperServers() { + if (this.dataZookeeperServers == null || this.dataZookeeperServers.isEmpty()) { + // If the configuration is not set, assuming that the dataZK is not enabled and all data is in the same + // ZooKeeper cluster + return this.getZookeeperServers(); + } + return dataZookeeperServers; + } + + public void setDataZookeeperServers(String dataZookeeperServers) { + this.dataZookeeperServers = dataZookeeperServers; + } + public String getGlobalZookeeperServers() { if (this.globalZookeeperServers == null || this.globalZookeeperServers.isEmpty()) { // If the configuration is not set, assuming that the globalZK is not enabled and all data is in the same @@ -818,6 +843,21 @@ public int getLoadBalancerNamespaceMaximumBundles() { return this.loadBalancerNamespaceMaximumBundles; } + public String getNamespaceOwnershipCacheFactoryProvider() { + return namespaceOwnershipCacheFactoryProvider; + } + + public void setNamespaceOwnershipCacheFactoryProvider(String namespaceOwnershipCacheFactoryProvider) { + this.namespaceOwnershipCacheFactoryProvider = namespaceOwnershipCacheFactoryProvider; + } + public String getLoadBalancerFactoryProvider() { + return loadBalancerFactoryProvider; + } + + public void setLoadBalancerFactoryProvider(String loadBalancerFactoryProvider) { + this.loadBalancerFactoryProvider = loadBalancerFactoryProvider; + } + public boolean isReplicationMetricsEnabled() { return replicationMetricsEnabled; } @@ -906,6 +946,14 @@ public void setZooKeeperSessionTimeoutMillis(long zooKeeperSessionTimeoutMillis) this.zooKeeperSessionTimeoutMillis = zooKeeperSessionTimeoutMillis; } + public long getDataZooKeeperSessionTimeoutMillis() { + return dataZooKeeperSessionTimeoutMillis; + } + + public void setDataZooKeeperSessionTimeoutMillis(long dataZooKeeperSessionTimeoutMillis) { + this.dataZooKeeperSessionTimeoutMillis = dataZooKeeperSessionTimeoutMillis; + } + public String getReplicatorPrefix() { return replicatorPrefix; } diff --git a/pulsar-broker/src/main/java/com/yahoo/pulsar/PulsarClusterMetadataSetup.java b/pulsar-broker/src/main/java/com/yahoo/pulsar/PulsarClusterMetadataSetup.java index 6730cfe9c55bb..42bd1da63da7f 100644 --- a/pulsar-broker/src/main/java/com/yahoo/pulsar/PulsarClusterMetadataSetup.java +++ b/pulsar-broker/src/main/java/com/yahoo/pulsar/PulsarClusterMetadataSetup.java @@ -21,6 +21,7 @@ import org.apache.bookkeeper.conf.ClientConfiguration; import org.apache.bookkeeper.meta.HierarchicalLedgerManagerFactory; import org.apache.bookkeeper.util.ZkUtils; +import static org.apache.commons.lang3.StringUtils.isBlank; import org.apache.zookeeper.CreateMode; import org.apache.zookeeper.ZooDefs; import org.apache.zookeeper.ZooKeeper; @@ -53,7 +54,11 @@ private static class Arguments { @Parameter(names = { "-zk", "--zookeeper" }, description = "Local ZooKeeper quorum connection string", required = true) - private String zookeeper; + private String localZookeeper; + + @Parameter(names = { "-dzk", + "--data-zookeeper" }, description = "Data ZooKeeper quorum connection string (Optional if data-zookeeper is not configured", required = false) + private String dataZookeeper; @Parameter(names = { "-gzk", "--global-zookeeper" }, description = "Global ZooKeeper quorum connection string", required = true) @@ -78,22 +83,26 @@ public static void main(String[] args) throws Exception { return; } - log.info("Setting up cluster {} with zk={} global-zk={}", arguments.cluster, arguments.zookeeper, - arguments.globalZookeeper); + arguments.dataZookeeper = isBlank(arguments.dataZookeeper) ? arguments.localZookeeper : arguments.dataZookeeper; + log.info("Setting up cluster {} with zk={} dzk={} global-zk={}", arguments.cluster, arguments.localZookeeper, + arguments.dataZookeeper, arguments.globalZookeeper); // Format BookKeeper metadata ClientConfiguration bkConf = new ClientConfiguration(); bkConf.setLedgerManagerFactoryClass(HierarchicalLedgerManagerFactory.class); - bkConf.setZkServers(arguments.zookeeper); + bkConf.setZkServers(arguments.dataZookeeper); if (!BookKeeperAdmin.format(bkConf, false /* interactive */, false /* force */)) { throw new IOException("Failed to initialize BookKeeper metadata"); } ZooKeeperClientFactory zkfactory = new ZookeeperClientFactoryImpl(); - ZooKeeper localZk = zkfactory.create(arguments.zookeeper, SessionType.ReadWrite, 30000).get(); + ZooKeeper localZk = zkfactory.create(arguments.localZookeeper, SessionType.ReadWrite, 30000).get(); + // use localZk for dataZookeeper if dataZk is not provided or it is same as localZk + ZooKeeper dataZk = arguments.localZookeeper.equals(arguments.dataZookeeper) ? localZk + : zkfactory.create(arguments.dataZookeeper, SessionType.ReadWrite, 30000).get(); ZooKeeper globalZk = zkfactory.create(arguments.globalZookeeper, SessionType.ReadWrite, 30000).get(); - localZk.create("/managed-ledgers", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); + dataZk.create("/managed-ledgers", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); localZk.create("/namespace", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); ZkUtils.createFullPathOptimistic(globalZk, "/admin/policies", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, diff --git a/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/MessagingServiceShutdownHook.java b/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/MessagingServiceShutdownHook.java index bba8026276c09..ad8fde8024b0e 100644 --- a/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/MessagingServiceShutdownHook.java +++ b/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/MessagingServiceShutdownHook.java @@ -83,8 +83,11 @@ public void shutdown(int exitCode) { try { // Try to close ZK session to ensure all ephemeral locks gets released immediately if (service != null) { - if (service.getZkClient().getState() != States.CLOSED) { - service.getZkClient().close(); + if (service.getLocalZkClient().getState() != States.CLOSED) { + service.getLocalZkClient().close(); + } + if (service.getDataZkClient().getState() != States.CLOSED) { + service.getDataZkClient().close(); } } } catch (Exception e) { diff --git a/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/PulsarService.java b/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/PulsarService.java index 218239a9dbf3a..1b2f36223f9b0 100644 --- a/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/PulsarService.java +++ b/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/PulsarService.java @@ -30,11 +30,13 @@ import org.apache.bookkeeper.mledger.ManagedLedgerFactory; import org.apache.bookkeeper.util.OrderedSafeExecutor; +import static org.apache.commons.lang3.StringUtils.isNotBlank; import org.apache.zookeeper.ZooKeeper; import org.eclipse.jetty.servlet.ServletHolder; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import com.google.common.annotations.VisibleForTesting; import com.google.common.collect.Lists; import com.yahoo.pulsar.broker.admin.AdminResource; import com.yahoo.pulsar.broker.cache.ConfigurationCacheService; @@ -42,6 +44,7 @@ import com.yahoo.pulsar.broker.loadbalance.LeaderElectionService; import com.yahoo.pulsar.broker.loadbalance.LeaderElectionService.LeaderListener; import com.yahoo.pulsar.broker.loadbalance.LoadManager; +import com.yahoo.pulsar.broker.loadbalance.LoadManagerFactory; import com.yahoo.pulsar.broker.loadbalance.LoadReportUpdaterTask; import com.yahoo.pulsar.broker.loadbalance.LoadResourceQuotaUpdaterTask; import com.yahoo.pulsar.broker.loadbalance.LoadSheddingTask; @@ -82,10 +85,13 @@ public class PulsarService implements AutoCloseable { private WebService webService = null; private WebSocketService webSocketService = null; private ConfigurationCacheService configurationCacheService = null; + private LocalZooKeeperCacheService dataZkCacheService = null; + private ZooKeeperCache dataZkCache; + private LocalZooKeeperConnectionService dataZooKeeperConnectionProvider; private LocalZooKeeperCacheService localZkCacheService = null; private ZooKeeperCache localZkCache; - private GlobalZooKeeperCache globalZkCache; private LocalZooKeeperConnectionService localZooKeeperConnectionProvider; + private GlobalZooKeeperCache globalZkCache; private final ScheduledExecutorService executor = Executors.newScheduledThreadPool(20, new DefaultThreadFactory("pulsar")); private final OrderedSafeExecutor orderedExecutor = new OrderedSafeExecutor(8, "pulsar-ordered"); @@ -171,12 +177,17 @@ public void close() throws PulsarServerException { globalZkCache.close(); globalZkCache = null; localZooKeeperConnectionProvider.close(); + dataZooKeeperConnectionProvider.close(); localZooKeeperConnectionProvider = null; + dataZooKeeperConnectionProvider = null; } configurationCacheService = null; + localZkCacheService = null; localZkCache = null; + dataZkCacheService = null; + dataZkCache = null; if (adminClient != null) { adminClient.close(); @@ -222,20 +233,34 @@ public void start() throws PulsarServerException { } // Now we are ready to start services - localZooKeeperConnectionProvider = new LocalZooKeeperConnectionService(getZooKeeperClientFactory(), + localZooKeeperConnectionProvider = new LocalZooKeeperConnectionService(getLocalZooKeeperClientFactory(), config.getZookeeperServers(), config.getZooKeeperSessionTimeoutMillis()); localZooKeeperConnectionProvider.start(shutdownService); - + + if (equalsDataAndLocalZk()) { + dataZooKeeperConnectionProvider = localZooKeeperConnectionProvider; + } else { + dataZooKeeperConnectionProvider = new LocalZooKeeperConnectionService(getZooKeeperClientFactory(), + config.getDataZookeeperServers(), config.getDataZooKeeperSessionTimeoutMillis()); + dataZooKeeperConnectionProvider.start(shutdownService); + } + // Initialize and start service to access configuration repository. this.startZkCacheService(); - managedLedgerClientFactory = new ManagedLedgerClientFactory(config, getZkClient(), + managedLedgerClientFactory = new ManagedLedgerClientFactory(config, getDataZkClient(), getBookKeeperClientFactory()); this.brokerService = new BrokerService(this); // Start load management service (even if load balancing is disabled) - this.loadManager = new SimpleLoadManagerImpl(this); + + if (isNotBlank(getConfiguration().getLoadBalancerFactoryProvider())) { + this.loadManager = ((LoadManagerFactory) Class.forName(getConfiguration().getLoadBalancerFactoryProvider()) + .newInstance()).create(this); + } else { + this.loadManager = new SimpleLoadManagerImpl(this); + } this.startLoadManagementService(); @@ -369,7 +394,8 @@ private void startZkCacheService() throws PulsarServerException { LOG.info("starting configuration cache service"); - this.localZkCache = new LocalZooKeeperCache(getZkClient(), getOrderedExecutor()); + this.localZkCache = new LocalZooKeeperCache(getLocalZkClient(), getOrderedExecutor()); + this.dataZkCache = equalsDataAndLocalZk() ? localZkCache : new LocalZooKeeperCache(getDataZkClient(), getOrderedExecutor()); this.globalZkCache = new GlobalZooKeeperCache(getZooKeeperClientFactory(), (int) config.getZooKeeperSessionTimeoutMillis(), config.getGlobalZookeeperServers(), getOrderedExecutor(), this.executor); @@ -381,6 +407,8 @@ private void startZkCacheService() throws PulsarServerException { this.configurationCacheService = new ConfigurationCacheService(getGlobalZkCache()); this.localZkCacheService = new LocalZooKeeperCacheService(getLocalZkCache(), this.configurationCacheService); + this.dataZkCacheService = equalsDataAndLocalZk() ? localZkCacheService + : new LocalZooKeeperCacheService(getDataZkCache(), this.configurationCacheService); } private void startNamespaceService() throws PulsarServerException { @@ -460,13 +488,21 @@ public String getStatusFilePath() { return config.getStatusFilePath(); } - public ZooKeeper getZkClient() { + public ZooKeeper getDataZkClient() { + return this.dataZooKeeperConnectionProvider.getLocalZooKeeper(); + } + + public ZooKeeper getLocalZkClient() { return this.localZooKeeperConnectionProvider.getLocalZooKeeper(); } public ConfigurationCacheService getConfigurationCache() { return configurationCacheService; } + + public boolean equalsDataAndLocalZk() { + return config.getZookeeperServers().equals(config.getDataZookeeperServers()); + } /** * Get the current pulsar state. @@ -508,6 +544,10 @@ public ManagedLedgerFactory getManagedLedgerFactory() { return managedLedgerClientFactory.getManagedLedgerFactory(); } + public ZooKeeperCache getDataZkCache() { + return dataZkCache; + } + public ZooKeeperCache getLocalZkCache() { return localZkCache; } @@ -528,10 +568,20 @@ public OrderedSafeExecutor getOrderedExecutor() { return orderedExecutor; } + public LocalZooKeeperCacheService getDataZkCacheService() { + return this.dataZkCacheService; + } + public LocalZooKeeperCacheService getLocalZkCacheService() { return this.localZkCacheService; } + // this is created to allow unit-test to mock it + @VisibleForTesting + public ZooKeeperClientFactory getLocalZooKeeperClientFactory() { + return getZooKeeperClientFactory(); + } + public ZooKeeperClientFactory getZooKeeperClientFactory() { if (zkClientFactory == null) { zkClientFactory = new ZookeeperClientFactoryImpl(); diff --git a/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/admin/AdminResource.java b/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/admin/AdminResource.java index ecf6923c03d1c..b980159f22db0 100644 --- a/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/admin/AdminResource.java +++ b/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/admin/AdminResource.java @@ -71,7 +71,7 @@ protected ZooKeeperCache globalZkCache() { } protected ZooKeeper localZk() { - return pulsar().getZkClient(); + return pulsar().getLocalZkClient(); } protected ZooKeeperCache localZkCache() { @@ -255,7 +255,7 @@ ZooKeeperDataCache clustersCache() { } ZooKeeperChildrenCache managedLedgerListCache() { - return pulsar().getLocalZkCacheService().managedLedgerListCache(); + return pulsar().getDataZkCacheService().managedLedgerListCache(); } Set clusters() { diff --git a/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/admin/Brokers.java b/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/admin/Brokers.java index bce449bddafba..134b8cae737f6 100644 --- a/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/admin/Brokers.java +++ b/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/admin/Brokers.java @@ -52,7 +52,7 @@ public Set getActiveBrokers(@PathParam("cluster") String cluster) throws try { // Add Native brokers - return pulsar().getLocalZkCache().getChildren(SimpleLoadManagerImpl.LOADBALANCE_BROKERS_ROOT); + return pulsar().getLoadManager().getActiveBrokers(); } catch (Exception e) { LOG.error(String.format("[%s] Failed to get active broker list: cluster=%s", clientAppId(), cluster), e); throw new RestException(e); diff --git a/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/cache/LocalZooKeeperCacheService.java b/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/cache/LocalZooKeeperCacheService.java index 3672374760a17..c255c59614575 100644 --- a/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/cache/LocalZooKeeperCacheService.java +++ b/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/cache/LocalZooKeeperCacheService.java @@ -43,7 +43,7 @@ public class LocalZooKeeperCacheService { private static final Logger LOG = LoggerFactory.getLogger(LocalZooKeeperCacheService.class); - private static final String MANAGED_LEDGER_ROOT = "/managed-ledgers"; + public static final String MANAGED_LEDGER_ROOT = "/managed-ledgers"; public static final String OWNER_INFO_ROOT = "/namespace"; public static final String LOCAL_POLICIES_ROOT = "/admin/local-policies"; diff --git a/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/loadbalance/LeaderElectionService.java b/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/loadbalance/LeaderElectionService.java index 4802403556d4e..66a7feb533dc7 100644 --- a/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/loadbalance/LeaderElectionService.java +++ b/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/loadbalance/LeaderElectionService.java @@ -71,7 +71,7 @@ public static interface LeaderListener { public LeaderElectionService(PulsarService pulsar, LeaderListener leaderListener) { this.pulsar = pulsar; - this.zkClient = pulsar.getZkClient(); + this.zkClient = pulsar.getLocalZkClient(); this.executor = pulsar.getExecutor(); this.leaderListener = leaderListener; this.jsonMapper = new ObjectMapper(); diff --git a/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/loadbalance/LoadManager.java b/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/loadbalance/LoadManager.java index ce6ef5666c782..7ee193162b4be 100644 --- a/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/loadbalance/LoadManager.java +++ b/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/loadbalance/LoadManager.java @@ -16,6 +16,8 @@ package com.yahoo.pulsar.broker.loadbalance; import java.util.List; +import java.util.Set; +import java.util.concurrent.CompletableFuture; import com.yahoo.pulsar.broker.PulsarServerException; import com.yahoo.pulsar.broker.stats.Metrics; @@ -47,6 +49,24 @@ public interface LoadManager { * Generate the load report */ LoadReport generateLoadReport() throws Exception; + + /** + * Get {@link LoadReport} of a broker + * + * @param brokerPath + * @return CompletableFuture completes with exception: + * org.apache.zookeeper.KeeperException.NoNodeException if {@link LoadReport} not found + * + */ + public CompletableFuture getLoadReportAsync(String broker); + + /** + * Returns list of active brokers which are registered in load-balancer + * + * @return + * @throws Exception + */ + Set getActiveBrokers() throws Exception; /** * Set flag to force load report update diff --git a/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/loadbalance/LoadManagerFactory.java b/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/loadbalance/LoadManagerFactory.java new file mode 100644 index 0000000000000..ddbdf1c767538 --- /dev/null +++ b/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/loadbalance/LoadManagerFactory.java @@ -0,0 +1,32 @@ +/** + * Copyright 2016 Yahoo Inc. + * + * Licensed 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 com.yahoo.pulsar.broker.loadbalance; + +import com.yahoo.pulsar.broker.PulsarService; +import com.yahoo.pulsar.broker.loadbalance.impl.SimpleLoadManagerDualZkImpl; + +public interface LoadManagerFactory { + + LoadManager create(PulsarService pulsar); + + public static class SimpleLoadManagerDualZkImplFactory implements LoadManagerFactory { + @Override + public LoadManager create(PulsarService pulsar) { + return new SimpleLoadManagerDualZkImpl(pulsar); + } + } + +} diff --git a/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/loadbalance/impl/SimpleLoadManagerDualZkImpl.java b/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/loadbalance/impl/SimpleLoadManagerDualZkImpl.java new file mode 100644 index 0000000000000..0b1b04a0f6d86 --- /dev/null +++ b/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/loadbalance/impl/SimpleLoadManagerDualZkImpl.java @@ -0,0 +1,154 @@ +/** + * Copyright 2016 Yahoo Inc. + * + * Licensed 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 com.yahoo.pulsar.broker.loadbalance.impl; + +import static org.apache.commons.lang3.StringUtils.isNotEmpty; + +import java.util.Set; +import java.util.concurrent.CompletableFuture; + +import org.apache.bookkeeper.util.ZkUtils; +import org.apache.zookeeper.CreateMode; +import org.apache.zookeeper.KeeperException; +import org.apache.zookeeper.ZooDefs.Ids; +import org.apache.zookeeper.data.Stat; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.google.common.base.Charsets; +import com.google.common.collect.Sets; +import com.yahoo.pulsar.broker.PulsarService; +import com.yahoo.pulsar.common.policies.data.loadbalancer.LoadReport; +import com.yahoo.pulsar.common.util.ObjectMapperFactory; +import com.yahoo.pulsar.zookeeper.ZooKeeperCacheListener; +import com.yahoo.pulsar.zookeeper.ZooKeeperChildrenCache; +import com.yahoo.pulsar.zookeeper.ZooKeeperDataCache; + +/** + *

+ * Sometimes if cluster has brokers which are running on different zk-quorum configuration.
+ * One set of brokers which are using single zk-quorom(data_zk) to store cluster-management and ledger-data. And other + * set of brokers are using separate zk: local_zk for cluster-management and data_zk for ledger-data. + *

+ * Therefore, loadBalancer requires loadReport + list_of_active brokers from both the cluster to decide assignment of + * new bundle-ownership. + *

+ * It helps to aggregate active brokers from local_zk and data_zk, also helps to get their load report by performing + * dual-read from both the cluster. + *

+ * As brokers which are running with one data_zk configuration, is not reading local_zk: So, this one also performs + * dual-write while broker registration and load-report generation so, it can be available at data_zk. + * + */ +public class SimpleLoadManagerDualZkImpl extends SimpleLoadManagerImpl { + + private final ZooKeeperChildrenCache _availableActiveBrokers; + private final ZooKeeperDataCache _loadReportCacheZk; + private final boolean shouldWriteOnDataZk; + + public SimpleLoadManagerDualZkImpl(PulsarService pulsar) { + super(pulsar); + _availableActiveBrokers = new ZooKeeperChildrenCache(pulsar.getDataZkCache(), LOADBALANCE_BROKERS_ROOT); + _availableActiveBrokers.registerListener(new ZooKeeperCacheListener>() { + @Override + public void onUpdate(String path, Set data, Stat stat) { + if (log.isDebugEnabled()) { + log.debug("Update Received for path {}", path); + } + scheduler.submit(SimpleLoadManagerDualZkImpl.super::updateRanking); + } + }); + _loadReportCacheZk = new ZooKeeperDataCache(pulsar.getDataZkCache()) { + @Override + public LoadReport deserialize(String key, byte[] content) throws Exception { + return ObjectMapperFactory.getThreadLocal().readValue(content, LoadReport.class); + } + }; + _loadReportCacheZk.registerListener(this); + shouldWriteOnDataZk = !pulsar.equalsDataAndLocalZk(); + } + + @Override + public Set getActiveBrokers() throws Exception { + Set availableBrokers = Sets.newHashSet(super.getActiveBrokers()); + availableBrokers.addAll(_availableActiveBrokers.get()); + return availableBrokers; + } + + @Override + public CompletableFuture getLoadReportAsync(String broker) { + CompletableFuture loadReportFuture = new CompletableFuture<>(); + super.getLoadReportAsync(broker).thenAccept(loadReport -> loadReportFuture.complete(loadReport)) + .exceptionally(ex -> { + String key = String.format("%s/%s", LOADBALANCE_BROKERS_ROOT, broker); + _loadReportCacheZk.getAsync(key).thenAccept(loadReport -> { + if (loadReport.isPresent()) { + loadReportFuture.complete(loadReport.get()); + } else { + loadReportFuture.completeExceptionally(new KeeperException.NoNodeException(key)); + } + }).exceptionally(e -> { + loadReportFuture.completeExceptionally(e); + return null; + }); + return null; + }); + return loadReportFuture; + } + + @Override + protected void registerBroker(String loadReportJson, String brokerZnodePath) + throws KeeperException, InterruptedException { + super.registerBroker(loadReportJson, brokerZnodePath); + if (shouldWriteOnDataZk) { + if (pulsar.getDataZkClient().exists(LOADBALANCE_BROKERS_ROOT, false) == null) { + try { + ZkUtils.createFullPathOptimistic(pulsar.getDataZkClient(), LOADBALANCE_BROKERS_ROOT, new byte[0], + Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); + } catch (KeeperException.NodeExistsException e) { + // ignore the exception, node might be present already + } + } + try { + ZkUtils.createFullPathOptimistic(pulsar.getDataZkClient(), brokerZnodePath, + loadReportJson.getBytes(Charsets.UTF_8), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL); + } catch (Exception e) { + // Catching excption here to print the right error message + log.error("Unable to create znode - [{}] for load balance on zookeeper ", brokerZnodePath, e); + throw e; + } + } + } + + @Override + public void disableBroker() throws Exception { + super.disableBroker(); + if (isNotEmpty(brokerZnodePath) && shouldWriteOnDataZk) { + pulsar.getDataZkClient().delete(brokerZnodePath, -1); + } + } + + @Override + protected void writeLoadReportOnZk(LoadReport lr) throws Exception { + super.writeLoadReportOnZk(lr); + if (shouldWriteOnDataZk) { + pulsar.getDataZkClient().setData(brokerZnodePath, + ObjectMapperFactory.getThreadLocal().writeValueAsBytes(lr), -1); + } + } + + private static final Logger log = LoggerFactory.getLogger(SimpleLoadManagerDualZkImpl.class); +} diff --git a/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/loadbalance/impl/SimpleLoadManagerImpl.java b/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/loadbalance/impl/SimpleLoadManagerImpl.java index fff4201d3636d..b70ec957464e2 100644 --- a/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/loadbalance/impl/SimpleLoadManagerImpl.java +++ b/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/loadbalance/impl/SimpleLoadManagerImpl.java @@ -31,6 +31,7 @@ import java.util.Map; import java.util.Set; import java.util.TreeMap; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; @@ -83,7 +84,7 @@ public class SimpleLoadManagerImpl implements LoadManager, ZooKeeperCacheListene private static final Logger log = LoggerFactory.getLogger(SimpleLoadManagerImpl.class); private final SimpleResourceAllocationPolicies policies; - private PulsarService pulsar; + protected PulsarService pulsar; // average JVM heap usage for private long avgJvmHeapUsageMBytes = 0; @@ -153,8 +154,8 @@ public class SimpleLoadManagerImpl implements LoadManager, ZooKeeperCacheListene public static final String LOADBALANCER_STRATEGY_LLS = "leastLoadedServer"; public static final String LOADBALANCER_STRATEGY_RAND = "weightedRandomSelection"; - private String brokerZnodePath; - private final ScheduledExecutorService scheduler; + protected String brokerZnodePath; + protected final ScheduledExecutorService scheduler; private final ZooKeeperChildrenCache availableActiveBrokers; private static final long MBytes = 1024 * 1024; @@ -235,14 +236,6 @@ public void start() throws PulsarServerException { try { // Register the brokers in zk list ServiceConfiguration conf = pulsar.getConfiguration(); - if (pulsar.getZkClient().exists(LOADBALANCE_BROKERS_ROOT, false) == null) { - try { - ZkUtils.createFullPathOptimistic(pulsar.getZkClient(), LOADBALANCE_BROKERS_ROOT, new byte[0], - Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); - } catch (KeeperException.NodeExistsException e) { - // ignore the exception, node might be present already - } - } String lookupServiceAddress = pulsar.getAdvertisedAddress() + ":" + conf.getWebServicePort(); brokerZnodePath = LOADBALANCE_BROKERS_ROOT + "/" + lookupServiceAddress; @@ -257,14 +250,8 @@ public void start() throws PulsarServerException { if (loadReport != null) { loadReportJson = ObjectMapperFactory.getThreadLocal().writeValueAsString(loadReport); } - try { - ZkUtils.createFullPathOptimistic(pulsar.getZkClient(), brokerZnodePath, - loadReportJson.getBytes(Charsets.UTF_8), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL); - } catch (Exception e) { - // Catching excption here to print the right error message - log.error("Unable to create znode - [{}] for load balance on zookeeper ", brokerZnodePath, e); - throw e; - } + registerBroker(loadReportJson, brokerZnodePath); + // first time, populate the broker ranking updateRanking(); log.info("Created broker ephemeral node on {}", brokerZnodePath); @@ -284,28 +271,43 @@ public void start() throws PulsarServerException { } } + protected void registerBroker(String loadReportJson, String brokerZnodePath) throws KeeperException, InterruptedException { + if (pulsar.getLocalZkClient().exists(LOADBALANCE_BROKERS_ROOT, false) == null) { + try { + ZkUtils.createFullPathOptimistic(pulsar.getLocalZkClient(), LOADBALANCE_BROKERS_ROOT, new byte[0], + Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); + } catch (KeeperException.NodeExistsException e) { + // ignore the exception, node might be present already + } + } + try { + ZkUtils.createFullPathOptimistic(pulsar.getLocalZkClient(), brokerZnodePath, + loadReportJson.getBytes(Charsets.UTF_8), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL); + } catch (Exception e) { + // Catching excption here to print the right error message + log.error("Unable to create znode - [{}] for load balance on zookeeper ", brokerZnodePath, e); + throw e; + } + } + @Override public void disableBroker() throws Exception { if (isNotEmpty(brokerZnodePath)) { - pulsar.getZkClient().delete(brokerZnodePath, -1); + pulsar.getLocalZkClient().delete(brokerZnodePath, -1); } } - public ZooKeeperChildrenCache getActiveBrokersCache() { - return this.availableActiveBrokers; - } - - public ZooKeeperDataCache getLoadReportCache() { - return this.loadReportCacheZk; + public Set getActiveBrokers() throws Exception { + return this.availableActiveBrokers.get(); } - + private void setDynamicConfigurationToZK(String zkPath, Map settings) throws IOException { byte[] settingBytes = ObjectMapperFactory.getThreadLocal().writeValueAsBytes(settings); try { if (pulsar.getLocalZkCache().exists(zkPath)) { - pulsar.getZkClient().setData(zkPath, settingBytes, -1); + pulsar.getLocalZkClient().setData(zkPath, settingBytes, -1); } else { - ZkUtils.createFullPathOptimistic(pulsar.getZkClient(), zkPath, settingBytes, Ids.OPEN_ACL_UNSAFE, + ZkUtils.createFullPathOptimistic(pulsar.getLocalZkClient(), zkPath, settingBytes, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); } } catch (Exception e) { @@ -943,7 +945,7 @@ private Map> getAvailableBrokers(ServiceUnitId serviceUn if (availableBrokers.isEmpty()) { // Create a map with all available brokers with no load information - Set activeBrokers = availableActiveBrokers.get(LOADBALANCE_BROKERS_ROOT); + Set activeBrokers = getActiveBrokers(); List brokersToShuffle = new ArrayList<>(activeBrokers); Collections.shuffle(brokersToShuffle); activeBrokers = new HashSet<>(brokersToShuffle); @@ -966,7 +968,7 @@ private ResourceUnit getLeastLoadedBroker(ServiceUnitId serviceUnit, // Remove candidates that point to inactive brokers Set activeBrokers = Collections.emptySet(); try { - activeBrokers = availableActiveBrokers.get(); + activeBrokers = getActiveBrokers(); // Need to use an explicit Iterator object to prevent concurrent modification exceptions Iterator> candidateIterator = finalCandidates.entries().iterator(); while (candidateIterator.hasNext()) { @@ -1008,16 +1010,14 @@ public void onUpdate(String path, LoadReport data, Stat stat) { scheduler.submit(this::updateRanking); } - private void updateRanking() { + protected void updateRanking() { try { synchronized (currentLoadReports) { currentLoadReports.clear(); - Set activeBrokers = availableActiveBrokers.get(); + Set activeBrokers = getActiveBrokers(); for (String broker : activeBrokers) { try { - String key = String.format("%s/%s", LOADBALANCE_BROKERS_ROOT, broker); - LoadReport lr = loadReportCacheZk.get(key) - .orElseThrow(() -> new KeeperException.NoNodeException()); + LoadReport lr = getLoadReport(broker); ResourceUnit ru = new SimpleResourceUnit(String.format("http://%s", lr.getName()), fromLoadReport(lr)); this.currentLoadReports.put(ru, lr); @@ -1108,6 +1108,27 @@ public LoadReport generateLoadReport() throws Exception { } } + private LoadReport getLoadReport(String broker) throws Exception { + return getLoadReportAsync(broker).get(); + } + + @Override + public CompletableFuture getLoadReportAsync(String broker) { + String key = String.format("%s/%s", LOADBALANCE_BROKERS_ROOT, broker); + CompletableFuture loadReportFuture = new CompletableFuture<>(); + loadReportCacheZk.getAsync(key).thenAccept(loadReport -> { + if (loadReport.isPresent()) { + loadReportFuture.complete(loadReport.get()); + } else { + loadReportFuture.completeExceptionally(new KeeperException.NoNodeException(key)); + } + }).exceptionally(ex -> { + loadReportFuture.completeExceptionally(ex); + return null; + }); + return loadReportFuture; + } + @Override public void setLoadReportForceUpdateFlag() { this.forceLoadReportUpdate = true; @@ -1195,8 +1216,7 @@ public void writeLoadReportOnZookeeper() throws Exception { if (needUpdate) { LoadReport lr = generateLoadReport(); - pulsar.getZkClient().setData(brokerZnodePath, ObjectMapperFactory.getThreadLocal().writeValueAsBytes(lr), - -1); + writeLoadReportOnZk(lr); this.lastLoadReport = lr; this.lastResourceUsageTimestamp = lr.getTimestamp(); // split-bundle if requires @@ -1204,6 +1224,11 @@ public void writeLoadReportOnZookeeper() throws Exception { } } + protected void writeLoadReportOnZk(LoadReport lr) throws Exception { + pulsar.getLocalZkClient().setData(brokerZnodePath, ObjectMapperFactory.getThreadLocal().writeValueAsBytes(lr), + -1); + } + private String getNamespaceNameFromBundleName(String bundleName) { // the bundle format is property/cluster/namespace/0x00000000_0xFFFFFFFF int pos = bundleName.lastIndexOf("/"); @@ -1410,4 +1435,5 @@ public void doNamespaceBundleSplit() throws Exception { public void stop() throws PulsarServerException { // do nothing } + } diff --git a/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/namespace/DualOwnershipCache.java b/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/namespace/DualOwnershipCache.java new file mode 100644 index 0000000000000..8eeb812256c1f --- /dev/null +++ b/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/namespace/DualOwnershipCache.java @@ -0,0 +1,220 @@ +/** + * Copyright 2016 Yahoo Inc. + * + * Licensed 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 com.yahoo.pulsar.broker.namespace; + +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; + +import org.apache.zookeeper.KeeperException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.github.benmanes.caffeine.cache.AsyncLoadingCache; +import com.google.common.collect.Lists; +import com.yahoo.pulsar.broker.PulsarService; +import com.yahoo.pulsar.client.util.FutureUtil; +import com.yahoo.pulsar.common.naming.NamespaceBundle; +import com.yahoo.pulsar.common.naming.NamespaceBundleFactory; +import com.yahoo.pulsar.common.naming.NamespaceBundles; +import com.yahoo.pulsar.zookeeper.ZooKeeperDataCache; + +import jersey.repackaged.com.google.common.collect.Maps; + +/** + * This {@link NamespaceOwnershipCache} helps to perform ownership-creation by creating znode on local_zk(primary-zk) + * and data_zk(secondary-zk) atomically. It also refers both the zk in order to find ownership of bundles. + *

+ * It is useful when cluster have two set of brokers (a) running with one zk instance (b) running with separate zk + * instance for cluster-management and ledger-data. + * + */ +public class DualOwnershipCache implements NamespaceOwnershipCache { + + private static final Logger LOG = LoggerFactory.getLogger(DualOwnershipCache.class); + + protected final OwnershipCacheImpl localZkOwnershipCache; + protected final OwnershipCacheImpl dataZkOwnershipCache; + private final boolean shouldWriteOnDataZk; + + /** + * Constructor of DualOwnershipCache + * + * @param ownerUrl + * the local broker URL that will be set as owner for the ServiceUnit + */ + public DualOwnershipCache(PulsarService pulsar, NamespaceBundleFactory bundleFactory) { + this.localZkOwnershipCache = new OwnershipCacheImpl(pulsar, pulsar.getLocalZkCache(), bundleFactory); + this.dataZkOwnershipCache = new OwnershipCacheImpl(pulsar, pulsar.getDataZkCache(), bundleFactory); + this.shouldWriteOnDataZk = !pulsar.equalsDataAndLocalZk(); + } + + /** + * Method to get the current owner of the ServiceUnit or set the local broker as the owner if absent + * + * @param suId + * identifier of the NamespaceBundle + * @return The ephemeral node data showing the current ownership info in ZooKeeper + * @throws Exception + */ + public CompletableFuture tryAcquiringOwnership(NamespaceBundle bundle) throws Exception { + final String path = ServiceUnitZkUtils.path(bundle); + + CompletableFuture future = new CompletableFuture<>(); + + LOG.info("Trying to acquire ownership of {}", bundle); + // Doing a get() on the ownedBundlesCache will trigger an async ZK write to acquire the lock over the + // service unit + localZkOwnershipCache.ownedBundlesCache.get(path).thenAccept(namespaceBundle -> { + LOG.info("Successfully acquired ownership of {}", path); + if (shouldWriteOnDataZk) { + // try to acquire ownership on secondary-zk + dataZkOwnershipCache.ownedBundlesCache.get(path).thenAccept(nsBundle -> { + LOG.info("Successfully acquired ownership of {} on secondary-zk", path); + future.complete(localZkOwnershipCache.selfOwnerInfo); + }).exceptionally(exception -> { + // as failed to take ownership of secondary-zk: remove ownership on primary-zk + localZkOwnershipCache.removeOwnership(bundle).handle((result, ex) -> { + if (ex != null) { + LOG.warn("Failed to remove ownership from primary-zk {}", path, ex); + } + handleOwnershipFailure(exception, path, bundle, dataZkOwnershipCache.ownershipReadOnlyCache, + dataZkOwnershipCache.ownedBundlesCache, future); + return null; + }); + ; + return null; + }); + } else { + future.complete(localZkOwnershipCache.selfOwnerInfo); + } + }).exceptionally(exception -> { + handleOwnershipFailure(exception, path, bundle, localZkOwnershipCache.ownershipReadOnlyCache, + localZkOwnershipCache.ownedBundlesCache, future); + return null; + }); + + return future; + } + + private static void handleOwnershipFailure(Throwable exception, final String path, NamespaceBundle bundle, + ZooKeeperDataCache readOnlyCache, + AsyncLoadingCache ownedBundlesCache, + CompletableFuture resultFuture) { + + // Failed to acquire ownership + if (exception instanceof CompletionException + && exception.getCause() instanceof KeeperException.NodeExistsException) { + LOG.info("Failed to acquire ownership of {} -- Already owned by other broker", path); + // Other broker acquired ownership at the same time, let's try to read it from the read-only cache + readOnlyCache.getAsync(path).thenAccept(ownerData -> { + if (LOG.isDebugEnabled()) { + LOG.debug("Found owner for {} at {}", bundle, ownerData); + } + + if (ownerData.isPresent()) { + resultFuture.complete(ownerData.get()); + } else { + // Strange scenario: we couldn't create a z-node because it was already existing, but when we + // try to read it, it's not there anymore + resultFuture.completeExceptionally(exception); + } + }).exceptionally(ex -> { + LOG.warn("Failed to check ownership of {}: {}", bundle, ex.getMessage(), ex); + resultFuture.completeExceptionally(exception); + return null; + }); + } else { + // Other ZK error, bailing out for now + LOG.warn("Failed to acquire ownership of {}: {}", bundle, exception.getMessage(), exception); + ownedBundlesCache.synchronous().invalidate(path); + resultFuture.completeExceptionally(exception); + } + } + + @Override + public CompletableFuture> getOwnerAsync(NamespaceBundle suname) { + CompletableFuture> result = new CompletableFuture<>(); + localZkOwnershipCache.getOwnerAsync(suname).thenApply(owner -> { + if (owner.isPresent()) { + result.complete(owner); + } else { + dataZkOwnershipCache.getOwnerAsync(suname).thenApply(secondary -> result.complete(secondary)) + .exceptionally(ex -> result.completeExceptionally(ex)); + } + return null; + }).exceptionally(ex -> result.completeExceptionally(ex)); + return result; + } + + @Override + public CompletableFuture removeOwnership(NamespaceBundle bundle) { + List> futures = Lists.newArrayList(); + futures.add(localZkOwnershipCache.removeOwnership(bundle)); + futures.add(dataZkOwnershipCache.removeOwnership(bundle)); + return FutureUtil.waitForAll(futures); + } + + @Override + public CompletableFuture removeOwnership(NamespaceBundles bundles) { + List> allFutures = Lists.newArrayList(); + allFutures.add(localZkOwnershipCache.removeOwnership(bundles)); + allFutures.add(dataZkOwnershipCache.removeOwnership(bundles)); + return FutureUtil.waitForAll(allFutures); + } + + @Override + public Map getOwnedBundles() { + Map ownedBundles = Maps + .newHashMap(localZkOwnershipCache.ownedBundlesCache.synchronous().asMap()); + ownedBundles.putAll(this.dataZkOwnershipCache.ownedBundlesCache.synchronous().asMap()); + return ownedBundles; + } + + @Override + public boolean isNamespaceBundleOwned(NamespaceBundle bundle) { + OwnedBundle primaryOwnedBundle = localZkOwnershipCache.getOwnedBundle(bundle); + OwnedBundle secondaryOwnedBundle = dataZkOwnershipCache.getOwnedBundle(bundle); + return (primaryOwnedBundle != null && primaryOwnedBundle.isActive()) + || (secondaryOwnedBundle != null && secondaryOwnedBundle.isActive()); + } + + @Override + public OwnedBundle getOwnedBundle(NamespaceBundle bundle) { + OwnedBundle ownedBundle = localZkOwnershipCache.getOwnedBundle(bundle); + return ownedBundle != null ? ownedBundle : dataZkOwnershipCache.getOwnedBundle(bundle); + } + + @Override + public void disableOwnership(NamespaceBundle bundle) throws Exception { + localZkOwnershipCache.disableOwnership(bundle); + dataZkOwnershipCache.disableOwnership(bundle); + } + + @Override + public void updateBundleState(NamespaceBundle bundle, boolean isActive) throws Exception { + localZkOwnershipCache.updateBundleState(bundle, isActive); + dataZkOwnershipCache.updateBundleState(bundle, isActive); + } + + @Override + public NamespaceEphemeralData getSelfOwnerInfo() { + return localZkOwnershipCache.getSelfOwnerInfo(); + } + +} diff --git a/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/namespace/NamespaceOwnershipCache.java b/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/namespace/NamespaceOwnershipCache.java new file mode 100644 index 0000000000000..e983f90c34b90 --- /dev/null +++ b/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/namespace/NamespaceOwnershipCache.java @@ -0,0 +1,116 @@ +/** + * Copyright 2016 Yahoo Inc. + * + * Licensed 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 com.yahoo.pulsar.broker.namespace; + +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.CompletableFuture; + +import com.yahoo.pulsar.broker.PulsarService; +import com.yahoo.pulsar.common.naming.NamespaceBundle; +import com.yahoo.pulsar.common.naming.NamespaceBundleFactory; +import com.yahoo.pulsar.common.naming.NamespaceBundles; + +/** + * This class provides a cache service for all the service unit ownership among the brokers. It provide a cache service + * as well as ZooKeeper read/write functions for a) lookup of a service unit ownership to a broker; b) take ownership of + * a service unit by the local broker + * + * + */ +public interface NamespaceOwnershipCache { + + /** + * Method to get the current owner of the ServiceUnit + * + * @param suId + * identifier of the ServiceUnit + * @return The ephemeral node data showing the current ownership info in ZooKeeper + * @throws Exception + * throws exception if no ownership info is found + */ + CompletableFuture> getOwnerAsync(NamespaceBundle suname); + + /** + * Method to get the current owner of the ServiceUnit or set the local broker as the owner if absent + * + * @param suId + * identifier of the NamespaceBundle + * @return The ephemeral node data showing the current ownership info in ZooKeeper + * @throws Exception + */ + CompletableFuture tryAcquiringOwnership(NamespaceBundle bundle) throws Exception; + + /** + * Method to remove the ownership of local broker on the NamespaceBundle, if owned + * + */ + CompletableFuture removeOwnership(NamespaceBundle bundle); + + /** + * Method to remove ownership of all owned bundles + * + * @param bundles + * NamespaceBundles to remove from ownership cache + */ + CompletableFuture removeOwnership(NamespaceBundles bundles); + + /** + * Method to access the map of all ServiceUnit objects owned by the local broker + * + * @return a map of owned ServiceUnit objects + */ + Map getOwnedBundles(); + + /** + * Checked whether a particular bundle is currently owned by this broker + * + * @param bundle + * @return + */ + boolean isNamespaceBundleOwned(NamespaceBundle bundle); + + /** + * Return the {@link OwnedBundle} instance from the local cache. Does not block. + * + * @param bundle + * @return + */ + OwnedBundle getOwnedBundle(NamespaceBundle bundle); + + /** + * Disable bundle in local cache and on zk + * + * @param bundle + * @throws Exception + */ + void disableOwnership(NamespaceBundle bundle) throws Exception; + + /** + * Update bundle state in a local cache + * + * @param bundle + * @throws Exception + */ + void updateBundleState(NamespaceBundle bundle, boolean isActive) throws Exception; + + /** + * Get Self-namespaceData node + * + * @return + */ + NamespaceEphemeralData getSelfOwnerInfo(); +} diff --git a/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/namespace/NamespaceService.java b/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/namespace/NamespaceService.java index 31c77090d6304..2e0e186aaf788 100644 --- a/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/namespace/NamespaceService.java +++ b/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/namespace/NamespaceService.java @@ -17,6 +17,7 @@ import static com.google.common.base.Preconditions.checkArgument; import static com.google.common.base.Preconditions.checkNotNull; +import static com.yahoo.pulsar.broker.admin.AdminResource.jsonMapper; import static com.yahoo.pulsar.broker.cache.LocalZooKeeperCacheService.LOCAL_POLICIES_ROOT; import static com.yahoo.pulsar.broker.web.PulsarWebResource.joinPath; import static com.yahoo.pulsar.common.naming.NamespaceBundleFactory.getBundlesData; @@ -49,8 +50,6 @@ import com.yahoo.pulsar.broker.ServiceConfiguration; import com.yahoo.pulsar.broker.admin.AdminResource; import com.yahoo.pulsar.broker.loadbalance.LoadManager; -import com.yahoo.pulsar.common.policies.data.loadbalancer.LoadReport; -import com.yahoo.pulsar.broker.loadbalance.impl.SimpleLoadManagerImpl; import com.yahoo.pulsar.broker.lookup.LookupResult; import com.yahoo.pulsar.broker.service.BrokerServiceException.ServiceUnitNotReadyException; import com.yahoo.pulsar.client.admin.PulsarAdmin; @@ -66,11 +65,11 @@ import com.yahoo.pulsar.common.policies.data.BundlesData; import com.yahoo.pulsar.common.policies.data.LocalPolicies; import com.yahoo.pulsar.common.policies.data.NamespaceOwnershipStatus; +import com.yahoo.pulsar.common.policies.data.loadbalancer.LoadReport; import com.yahoo.pulsar.common.policies.impl.NamespaceIsolationPolicies; import com.yahoo.pulsar.common.util.Codec; import com.yahoo.pulsar.common.util.ObjectMapperFactory; import com.yahoo.pulsar.zookeeper.ZooKeeperCache.Deserializer; -import static com.yahoo.pulsar.broker.admin.AdminResource.jsonMapper; /** * The NamespaceService provides resource ownership lookup as well as resource ownership claiming services @@ -97,7 +96,7 @@ public enum AddressType { private final PulsarService pulsar; - private final OwnershipCache ownershipCache; + private final NamespaceOwnershipCache ownershipCache; private final NamespaceBundleFactory bundleFactory; @@ -123,7 +122,18 @@ public NamespaceService(PulsarService pulsar) { this.loadManager = pulsar.getLoadManager(); ServiceUnitZkUtils.initZK(pulsar.getLocalZkCache().getZooKeeper(), pulsar.getBrokerServiceUrl()); this.bundleFactory = new NamespaceBundleFactory(pulsar, Hashing.crc32()); - this.ownershipCache = new OwnershipCache(pulsar, bundleFactory); + if (StringUtils.isNotBlank(pulsar.getConfiguration().getNamespaceOwnershipCacheFactoryProvider())) { + try { + this.ownershipCache = ((OwnershipCacheFactory) Class + .forName(pulsar.getConfiguration().getNamespaceOwnershipCacheFactoryProvider()) + .newInstance()).create(pulsar, bundleFactory); + } catch (Exception e) { + throw new IllegalArgumentException("Failed to initialize OwnershipCacheFactory " + + pulsar.getConfiguration().getNamespaceOwnershipCacheFactoryProvider()); + } + } else { + this.ownershipCache = new OwnershipCacheImpl(pulsar, pulsar.getLocalZkCache(), bundleFactory); + } } public CompletableFuture getBrokerServiceUrlAsync(DestinationName topic, boolean authoritative) { @@ -406,29 +416,23 @@ private CompletableFuture createLookupResult(String candidateBroke try { checkArgument(StringUtils.isNotBlank(candidateBroker), "Lookup broker can't be null " + candidateBroker); URI uri = new URI(candidateBroker); - String path = String.format("%s/%s:%s", SimpleLoadManagerImpl.LOADBALANCE_BROKERS_ROOT, uri.getHost(), - uri.getPort()); - pulsar.getLocalZkCache().getDataAsync(path, loadReportDeserializer).thenAccept(reportData -> { - if (reportData.isPresent()) { - LoadReport report = reportData.get(); - lookupFuture.complete(new LookupResult(report.getWebServiceUrl(), report.getWebServiceUrlTls(), - report.getPulsarServiceUrl(), report.getPulsarServieUrlTls())); - } else { - lookupFuture.completeExceptionally(new KeeperException.NoNodeException(path)); - } + String path = String.format("%s:%s", uri.getHost(), uri.getPort()); + pulsar.getLoadManager().getLoadReportAsync(path).thenAccept(report -> { + lookupFuture.complete(new LookupResult(report.getWebServiceUrl(), report.getWebServiceUrlTls(), + report.getPulsarServiceUrl(), report.getPulsarServieUrlTls())); }).exceptionally(ex -> { lookupFuture.completeExceptionally(ex); return null; }); + } catch (Exception e) { lookupFuture.completeExceptionally(e); } return lookupFuture; } - private boolean isBrokerActive(String candidateBroker) throws KeeperException, InterruptedException { - Set activeNativeBrokers = pulsar.getLocalZkCache() - .getChildren(SimpleLoadManagerImpl.LOADBALANCE_BROKERS_ROOT); + private boolean isBrokerActive(String candidateBroker) throws Exception { + Set activeNativeBrokers = pulsar.getLoadManager().getActiveBrokers(); for (String brokerHostPort : activeNativeBrokers) { if (candidateBroker.equals("http://" + brokerHostPort)) { @@ -626,7 +630,7 @@ private void updateNamespaceBundles(NamespaceName nsname, NamespaceBundles nsBun ObjectMapperFactory.getThreadLocal().writeValueAsBytes(policies.get()), -1, callback, null); } - public OwnershipCache getOwnershipCache() { + public NamespaceOwnershipCache getOwnershipCache() { return ownershipCache; } @@ -703,7 +707,7 @@ public List getListOfDestinations(String property, String cluster, Strin try { String path = String.format("/managed-ledgers/%s/%s/%s/persistent", property, cluster, namespace); LOG.debug("Getting children from managed-ledgers now: {}", path); - for (String destination : pulsar.getLocalZkCacheService().managedLedgerListCache().get(path)) { + for (String destination : pulsar.getDataZkCacheService().managedLedgerListCache().get(path)) { destinations.add(String.format("persistent://%s/%s/%s/%s", property, cluster, namespace, Codec.decode(destination))); } diff --git a/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/namespace/OwnershipCacheFactory.java b/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/namespace/OwnershipCacheFactory.java new file mode 100644 index 0000000000000..864ccbaed2ef2 --- /dev/null +++ b/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/namespace/OwnershipCacheFactory.java @@ -0,0 +1,34 @@ +/** + * Copyright 2016 Yahoo Inc. + * + * Licensed 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 com.yahoo.pulsar.broker.namespace; + +import com.yahoo.pulsar.broker.PulsarService; +import com.yahoo.pulsar.common.naming.NamespaceBundleFactory; + +public interface OwnershipCacheFactory { + + NamespaceOwnershipCache create(PulsarService pulsar, NamespaceBundleFactory bundleFactory); + + public static class OwnershipDualCacheFactoryImpl implements OwnershipCacheFactory { + + @Override + public NamespaceOwnershipCache create(PulsarService pulsar, NamespaceBundleFactory bundleFactory) { + return new DualOwnershipCache(pulsar, bundleFactory); + } + + } + +} diff --git a/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/namespace/OwnershipCache.java b/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/namespace/OwnershipCacheImpl.java similarity index 96% rename from pulsar-broker/src/main/java/com/yahoo/pulsar/broker/namespace/OwnershipCache.java rename to pulsar-broker/src/main/java/com/yahoo/pulsar/broker/namespace/OwnershipCacheImpl.java index f2e4a359ae070..875c8c9004c70 100644 --- a/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/namespace/OwnershipCache.java +++ b/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/namespace/OwnershipCacheImpl.java @@ -15,8 +15,6 @@ */ package com.yahoo.pulsar.broker.namespace; -import static com.google.common.base.Preconditions.checkState; - import java.util.List; import java.util.Map; import java.util.Optional; @@ -36,8 +34,6 @@ import com.github.benmanes.caffeine.cache.AsyncCacheLoader; import com.github.benmanes.caffeine.cache.AsyncLoadingCache; import com.github.benmanes.caffeine.cache.Caffeine; -import com.github.benmanes.caffeine.cache.RemovalCause; -import com.github.benmanes.caffeine.cache.RemovalListener; import com.google.common.collect.Lists; import com.google.common.util.concurrent.MoreExecutors; import com.yahoo.pulsar.broker.PulsarService; @@ -56,9 +52,9 @@ * * */ -public class OwnershipCache { +public class OwnershipCacheImpl implements NamespaceOwnershipCache{ - private static final Logger LOG = LoggerFactory.getLogger(OwnershipCache.class); + private static final Logger LOG = LoggerFactory.getLogger(OwnershipCacheImpl.class); /** * The local broker URL that this OwnershipCache will set as owner @@ -73,7 +69,7 @@ public class OwnershipCache { /** * The NamespaceEphemeralData objects that can be associated with the current owner */ - private final NamespaceEphemeralData selfOwnerInfo; + protected final NamespaceEphemeralData selfOwnerInfo; /** * The NamespaceEphemeralData objects that can be associated with the current owner, when the broker is disabled. @@ -84,12 +80,12 @@ public class OwnershipCache { * Service unit ownership cache of ZooKeeper data of ephemeral nodes showing all known ownership of * service unit to active brokers */ - private final ZooKeeperDataCache ownershipReadOnlyCache; + protected final ZooKeeperDataCache ownershipReadOnlyCache; /** * The loading cache of locally owned NamespaceBundle objects */ - private final AsyncLoadingCache ownedBundlesCache; + protected final AsyncLoadingCache ownedBundlesCache; /** * The ObjectMapper to deserialize/serialize JSON objects @@ -149,7 +145,7 @@ public CompletableFuture asyncLoad(String namespaceBundleZNode, Exe * @param ownerUrl * the local broker URL that will be set as owner for the ServiceUnit */ - public OwnershipCache(PulsarService pulsar, NamespaceBundleFactory bundleFactory) { + public OwnershipCacheImpl(PulsarService pulsar, ZooKeeperCache localZkCache, NamespaceBundleFactory bundleFactory) { this.ownerBrokerUrl = pulsar.getBrokerServiceUrl(); this.ownerBrokerUrlTls = pulsar.getBrokerServiceUrlTls(); this.selfOwnerInfo = new NamespaceEphemeralData(ownerBrokerUrl, ownerBrokerUrlTls, @@ -157,7 +153,7 @@ public OwnershipCache(PulsarService pulsar, NamespaceBundleFactory bundleFactory this.selfOwnerInfoDisabled = new NamespaceEphemeralData(ownerBrokerUrl, ownerBrokerUrlTls, pulsar.getWebServiceAddress(), pulsar.getWebServiceAddressTls(), true); this.bundleFactory = bundleFactory; - this.localZkCache = pulsar.getLocalZkCache(); + this.localZkCache = localZkCache; this.ownershipReadOnlyCache = pulsar.getLocalZkCacheService().ownerInfoCache(); // ownedBundlesCache contains all namespaces that are owned by the local broker this.ownedBundlesCache = Caffeine.newBuilder().executor(MoreExecutors.sameThreadExecutor()) diff --git a/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/service/BrokerService.java index c999790df3958..1ea533486f8f0 100644 --- a/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/service/BrokerService.java +++ b/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/service/BrokerService.java @@ -191,7 +191,7 @@ public BrokerService(PulsarService pulsar) throws Exception { } public void start() throws Exception { - this.producerNameGenerator = new DistributedIdGenerator(pulsar.getZkClient(), producerNameGeneratorPath, + this.producerNameGenerator = new DistributedIdGenerator(pulsar.getLocalZkClient(), producerNameGeneratorPath, pulsar.getConfiguration().getClusterName()); ServerBootstrap bootstrap = new ServerBootstrap(); diff --git a/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/admin/AdminTest.java b/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/admin/AdminTest.java index 781d028e12a7c..ca6c115db2657 100644 --- a/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/admin/AdminTest.java +++ b/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/admin/AdminTest.java @@ -102,7 +102,7 @@ public void setup() throws Exception { clusters = spy(new Clusters()); clusters.setPulsar(pulsar); - doReturn(mockZookKeeper).when(clusters).globalZk(); + doReturn(dataMockZookKeeper).when(clusters).globalZk(); doReturn(configurationCache.clustersCache()).when(clusters).clustersCache(); doReturn(configurationCache.clustersListCache()).when(clusters).clustersListCache(); doReturn(configurationCache.namespaceIsolationPoliciesCache()).when(clusters).namespaceIsolationPoliciesCache(); @@ -112,7 +112,7 @@ public void setup() throws Exception { properties = spy(new Properties()); properties.setServletContext(new MockServletContext()); properties.setPulsar(pulsar); - doReturn(mockZookKeeper).when(properties).globalZk(); + doReturn(dataMockZookKeeper).when(properties).globalZk(); doReturn(configurationCache.propertiesCache()).when(properties).propertiesCache(); doReturn("test").when(properties).clientAppId(); doNothing().when(properties).validateSuperUserAccess(); @@ -120,8 +120,8 @@ public void setup() throws Exception { namespaces = spy(new Namespaces()); namespaces.setServletContext(new MockServletContext()); namespaces.setPulsar(pulsar); - doReturn(mockZookKeeper).when(namespaces).globalZk(); - doReturn(mockZookKeeper).when(namespaces).localZk(); + doReturn(dataMockZookKeeper).when(namespaces).globalZk(); + doReturn(dataMockZookKeeper).when(namespaces).localZk(); doReturn(configurationCache.propertiesCache()).when(namespaces).propertiesCache(); doReturn(configurationCache.policiesCache()).when(namespaces).policiesCache(); doReturn("test").when(namespaces).clientAppId(); @@ -133,8 +133,8 @@ public void setup() throws Exception { brokers = spy(new Brokers()); brokers.setServletContext(new MockServletContext()); brokers.setPulsar(pulsar); - doReturn(mockZookKeeper).when(brokers).globalZk(); - doReturn(mockZookKeeper).when(brokers).localZk(); + doReturn(dataMockZookKeeper).when(brokers).globalZk(); + doReturn(dataMockZookKeeper).when(brokers).localZk(); doReturn(configurationCache.clustersListCache()).when(brokers).clustersListCache(); doReturn("test").when(brokers).clientAppId(); doNothing().when(brokers).validateSuperUserAccess(); @@ -146,8 +146,8 @@ public void setup() throws Exception { persistentTopics = spy(new PersistentTopics()); persistentTopics.setServletContext(new MockServletContext()); persistentTopics.setPulsar(pulsar); - doReturn(mockZookKeeper).when(persistentTopics).globalZk(); - doReturn(mockZookKeeper).when(persistentTopics).localZk(); + doReturn(dataMockZookKeeper).when(persistentTopics).globalZk(); + doReturn(dataMockZookKeeper).when(persistentTopics).localZk(); doReturn(configurationCache.propertiesCache()).when(persistentTopics).propertiesCache(); doReturn(configurationCache.policiesCache()).when(persistentTopics).policiesCache(); doReturn("test").when(persistentTopics).clientAppId(); @@ -159,16 +159,16 @@ public void setup() throws Exception { resourceQuotas = spy(new ResourceQuotas()); resourceQuotas.setServletContext(new MockServletContext()); resourceQuotas.setPulsar(pulsar); - doReturn(mockZookKeeper).when(resourceQuotas).globalZk(); - doReturn(mockZookKeeper).when(resourceQuotas).localZk(); + doReturn(dataMockZookKeeper).when(resourceQuotas).globalZk(); + doReturn(dataMockZookKeeper).when(resourceQuotas).localZk(); doReturn(configurationCache.propertiesCache()).when(resourceQuotas).propertiesCache(); doReturn(configurationCache.policiesCache()).when(resourceQuotas).policiesCache(); brokerStats = spy(new BrokerStats()); brokerStats.setServletContext(new MockServletContext()); brokerStats.setPulsar(pulsar); - doReturn(mockZookKeeper).when(brokerStats).globalZk(); - doReturn(mockZookKeeper).when(brokerStats).localZk(); + doReturn(dataMockZookKeeper).when(brokerStats).globalZk(); + doReturn(dataMockZookKeeper).when(brokerStats).localZk(); doReturn(configurationCache.propertiesCache()).when(brokerStats).propertiesCache(); doReturn(configurationCache.policiesCache()).when(brokerStats).policiesCache(); } @@ -271,7 +271,7 @@ void clusters() throws Exception { } // Test zk failures - mockZookKeeper.failNow(Code.SESSIONEXPIRED); + dataMockZookKeeper.failNow(Code.SESSIONEXPIRED); configurationCache.clustersListCache().clear(); try { clusters.getClusters(); @@ -280,7 +280,7 @@ void clusters() throws Exception { assertEquals(e.getResponse().getStatus(), Status.INTERNAL_SERVER_ERROR.getStatusCode()); } - mockZookKeeper.failNow(Code.SESSIONEXPIRED); + dataMockZookKeeper.failNow(Code.SESSIONEXPIRED); try { clusters.createCluster("test", new ClusterData("http://broker.messaging.test.example.com")); fail("should have failed"); @@ -288,7 +288,7 @@ void clusters() throws Exception { assertEquals(e.getResponse().getStatus(), Status.INTERNAL_SERVER_ERROR.getStatusCode()); } - mockZookKeeper.failNow(Code.SESSIONEXPIRED); + dataMockZookKeeper.failNow(Code.SESSIONEXPIRED); try { clusters.updateCluster("test", new ClusterData("http://broker.messaging.test.example.com")); fail("should have failed"); @@ -296,7 +296,7 @@ void clusters() throws Exception { assertEquals(e.getResponse().getStatus(), Status.INTERNAL_SERVER_ERROR.getStatusCode()); } - mockZookKeeper.failNow(Code.SESSIONEXPIRED); + dataMockZookKeeper.failNow(Code.SESSIONEXPIRED); try { clusters.getCluster("test"); fail("should have failed"); @@ -304,7 +304,7 @@ void clusters() throws Exception { assertEquals(e.getResponse().getStatus(), Status.INTERNAL_SERVER_ERROR.getStatusCode()); } - mockZookKeeper.failAfter(0, Code.SESSIONEXPIRED); + dataMockZookKeeper.failAfter(0, Code.SESSIONEXPIRED); try { clusters.deleteCluster("use"); fail("should have failed"); @@ -312,7 +312,7 @@ void clusters() throws Exception { assertEquals(e.getResponse().getStatus(), Status.INTERNAL_SERVER_ERROR.getStatusCode()); } - mockZookKeeper.failAfter(1, Code.SESSIONEXPIRED); + dataMockZookKeeper.failAfter(1, Code.SESSIONEXPIRED); try { clusters.deleteCluster("use"); fail("should have failed"); @@ -388,7 +388,7 @@ void properties() throws Exception { } // Test zk failures - mockZookKeeper.failNow(Code.SESSIONEXPIRED); + dataMockZookKeeper.failNow(Code.SESSIONEXPIRED); try { properties.getProperties(); fail("should have failed"); @@ -396,7 +396,7 @@ void properties() throws Exception { assertEquals(e.getResponse().getStatus(), Status.INTERNAL_SERVER_ERROR.getStatusCode()); } - mockZookKeeper.failNow(Code.SESSIONEXPIRED); + dataMockZookKeeper.failNow(Code.SESSIONEXPIRED); try { properties.getPropertyAdmin("my-property"); fail("should have failed"); @@ -404,7 +404,7 @@ void properties() throws Exception { assertEquals(e.getResponse().getStatus(), Status.INTERNAL_SERVER_ERROR.getStatusCode()); } - mockZookKeeper.failNow(Code.SESSIONEXPIRED); + dataMockZookKeeper.failNow(Code.SESSIONEXPIRED); try { properties.updateProperty("my-property", newPropertyAdmin); fail("should have failed"); @@ -412,7 +412,7 @@ void properties() throws Exception { assertEquals(e.getResponse().getStatus(), Status.INTERNAL_SERVER_ERROR.getStatusCode()); } - mockZookKeeper.failNow(Code.SESSIONEXPIRED); + dataMockZookKeeper.failNow(Code.SESSIONEXPIRED); try { properties.createProperty("test", propertyAdmin); fail("should have failed"); @@ -420,7 +420,7 @@ void properties() throws Exception { assertEquals(e.getResponse().getStatus(), Status.INTERNAL_SERVER_ERROR.getStatusCode()); } - mockZookKeeper.failNow(Code.SESSIONEXPIRED); + dataMockZookKeeper.failNow(Code.SESSIONEXPIRED); try { properties.deleteProperty("my-property"); fail("should have failed"); @@ -429,7 +429,7 @@ void properties() throws Exception { } properties.createProperty("error-property", propertyAdmin); - mockZookKeeper.failAfter(2, Code.SESSIONEXPIRED); + dataMockZookKeeper.failAfter(2, Code.SESSIONEXPIRED); try { properties.deleteProperty("error-property"); fail("should have failed"); @@ -525,7 +525,7 @@ void resourceQuotas() throws Exception { // create policies PropertyAdmin admin = new PropertyAdmin(); admin.getAllowedClusters().add(cluster); - mockZookKeeper.create(PulsarWebResource.path("policies", property), + dataMockZookKeeper.create(PulsarWebResource.path("policies", property), ObjectMapperFactory.getThreadLocal().writeValueAsBytes(admin), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); @@ -581,7 +581,7 @@ void persistentTopics() throws Exception { // create policies PropertyAdmin admin = new PropertyAdmin(); admin.getAllowedClusters().add(cluster); - ZkUtils.createFullPathOptimistic(mockZookKeeper, + ZkUtils.createFullPathOptimistic(dataMockZookKeeper, PulsarWebResource.path("policies", property, cluster, namespace), ObjectMapperFactory.getThreadLocal().writeValueAsBytes(new Policies()), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); diff --git a/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/admin/NamespacesTest.java b/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/admin/NamespacesTest.java index 54f50ffa9077a..515f352e6f8b8 100644 --- a/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/admin/NamespacesTest.java +++ b/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/admin/NamespacesTest.java @@ -59,7 +59,7 @@ import com.yahoo.pulsar.broker.auth.MockedPulsarServiceBaseTest; import com.yahoo.pulsar.broker.namespace.NamespaceEphemeralData; import com.yahoo.pulsar.broker.namespace.NamespaceService; -import com.yahoo.pulsar.broker.namespace.OwnershipCache; +import com.yahoo.pulsar.broker.namespace.NamespaceOwnershipCache; import com.yahoo.pulsar.broker.web.PulsarWebResource; import com.yahoo.pulsar.broker.web.RestException; import com.yahoo.pulsar.client.admin.PulsarAdminException; @@ -119,8 +119,8 @@ public void setup() throws Exception { namespaces = spy(new Namespaces()); namespaces.setServletContext(new MockServletContext()); namespaces.setPulsar(pulsar); - doReturn(mockZookKeeper).when(namespaces).globalZk(); - doReturn(mockZookKeeper).when(namespaces).localZk(); + doReturn(dataMockZookKeeper).when(namespaces).globalZk(); + doReturn(dataMockZookKeeper).when(namespaces).localZk(); doReturn(pulsar.getConfigurationCache().propertiesCache()).when(namespaces).propertiesCache(); doReturn(pulsar.getConfigurationCache().policiesCache()).when(namespaces).policiesCache(); doReturn(false).when(namespaces).isRequestHttps(); @@ -186,7 +186,7 @@ public void testCreateNamespaces() throws Exception { assertEquals(e.getResponse().getStatus(), Status.PRECONDITION_FAILED.getStatusCode()); } - mockZookKeeper.failNow(Code.SESSIONEXPIRED); + dataMockZookKeeper.failNow(Code.SESSIONEXPIRED); try { namespaces.createNamespace("my-property", "use", "my-namespace-3", new BundlesData()); fail("should have failed"); @@ -222,7 +222,7 @@ public void testGetNamespaces() throws Exception { } // ZK Errors - mockZookKeeper.failNow(Code.SESSIONEXPIRED); + dataMockZookKeeper.failNow(Code.SESSIONEXPIRED); try { namespaces.getPropertyNamespaces(this.testProperty); fail("should have failed"); @@ -230,7 +230,7 @@ public void testGetNamespaces() throws Exception { // Ok } - mockZookKeeper.failNow(Code.SESSIONEXPIRED); + dataMockZookKeeper.failNow(Code.SESSIONEXPIRED); try { namespaces.getNamespacesForCluster(this.testProperty, this.testLocalCluster); fail("should have failed"); @@ -306,7 +306,7 @@ public void testGrantAndRevokePermissions() throws Exception { NamespaceName testNs = this.testLocalNamespaces.get(1); - mockZookKeeper.failNow(Code.SESSIONEXPIRED); + dataMockZookKeeper.failNow(Code.SESSIONEXPIRED); try { namespaces.getPolicies(testNs.getProperty(), testNs.getCluster(), testNs.getLocalName()); fail("should have failed"); @@ -314,7 +314,7 @@ public void testGrantAndRevokePermissions() throws Exception { // Ok } - mockZookKeeper.failNow(Code.SESSIONEXPIRED); + dataMockZookKeeper.failNow(Code.SESSIONEXPIRED); try { namespaces.getPermissions(testNs.getProperty(), testNs.getCluster(), testNs.getLocalName()); fail("should have failed"); @@ -322,7 +322,7 @@ public void testGrantAndRevokePermissions() throws Exception { // Ok } - mockZookKeeper.failNow(Code.SESSIONEXPIRED); + dataMockZookKeeper.failNow(Code.SESSIONEXPIRED); try { namespaces.grantPermissionOnNamespace(testNs.getProperty(), testNs.getCluster(), testNs.getLocalName(), "other-role", EnumSet.of(AuthAction.consume)); @@ -331,7 +331,7 @@ public void testGrantAndRevokePermissions() throws Exception { // Ok } - mockZookKeeper.failNow(Code.BADVERSION); + dataMockZookKeeper.failNow(Code.BADVERSION); try { namespaces.grantPermissionOnNamespace(testNs.getProperty(), testNs.getCluster(), testNs.getLocalName(), "other-role", EnumSet.of(AuthAction.consume)); @@ -340,7 +340,7 @@ public void testGrantAndRevokePermissions() throws Exception { assertEquals(e.getResponse().getStatus(), Status.CONFLICT.getStatusCode()); } - mockZookKeeper.failNow(Code.BADVERSION); + dataMockZookKeeper.failNow(Code.BADVERSION); try { namespaces.revokePermissionsOnNamespace(testNs.getProperty(), testNs.getCluster(), testNs.getLocalName(), "other-role"); @@ -349,7 +349,7 @@ public void testGrantAndRevokePermissions() throws Exception { assertEquals(e.getResponse().getStatus(), Status.CONFLICT.getStatusCode()); } - mockZookKeeper.failNow(Code.SESSIONEXPIRED); + dataMockZookKeeper.failNow(Code.SESSIONEXPIRED); try { namespaces.revokePermissionsOnNamespace(testNs.getProperty(), testNs.getCluster(), testNs.getLocalName(), "other-role"); @@ -416,7 +416,7 @@ public void testGlobalNamespaceReplicationConfiguration() throws Exception { // Sometimes watcher event consumes scheduled exception, so set to always fail to ensure exception is // thrown for api call. - mockZookKeeper.setAlwaysFail(Code.SESSIONEXPIRED); + dataMockZookKeeper.setAlwaysFail(Code.SESSIONEXPIRED); pulsar.getConfigurationCache().policiesCache().invalidate(AdminResource.path("policies", this.testProperty, "global", this.testGlobalNamespaces.get(0).getLocalName())); try { @@ -426,10 +426,10 @@ public void testGlobalNamespaceReplicationConfiguration() throws Exception { } catch (RestException e) { assertEquals(e.getResponse().getStatus(), Status.INTERNAL_SERVER_ERROR.getStatusCode()); } finally { - mockZookKeeper.unsetAlwaysFail(); + dataMockZookKeeper.unsetAlwaysFail(); } - mockZookKeeper.failNow(Code.BADVERSION); + dataMockZookKeeper.failNow(Code.BADVERSION); try { namespaces.setNamespaceReplicationClusters(this.testProperty, "global", this.testGlobalNamespaces.get(0).getLocalName(), Lists.newArrayList("use")); @@ -453,7 +453,7 @@ public void testGlobalNamespaceReplicationConfiguration() throws Exception { assertEquals(e.getResponse().getStatus(), Status.NOT_FOUND.getStatusCode()); } - mockZookKeeper.failNow(Code.SESSIONEXPIRED); + dataMockZookKeeper.failNow(Code.SESSIONEXPIRED); pulsar.getConfigurationCache().policiesCache().clear(); // ensure the ZooKeeper read happens, bypassing the cache @@ -592,7 +592,7 @@ public void testDeleteNamespaces() throws Exception { NamespaceName testNs = this.testLocalNamespaces.get(1); DestinationName topicName = DestinationName.get(testNs.getPersistentTopicName("my-topic")); - ZkUtils.createFullPathOptimistic(mockZookKeeper, "/managed-ledgers/" + topicName.getPersistenceNamingEncoding(), + ZkUtils.createFullPathOptimistic(dataMockZookKeeper, "/managed-ledgers/" + topicName.getPersistenceNamingEncoding(), new byte[0], null, null); // setup ownership to localhost @@ -632,9 +632,9 @@ public void testDeleteNamespaces() throws Exception { } // delete the topic from ZK - mockZookKeeper.delete("/managed-ledgers/" + topicName.getPersistenceNamingEncoding(), -1); + dataMockZookKeeper.delete("/managed-ledgers/" + topicName.getPersistenceNamingEncoding(), -1); // ensure refreshed destination list in the cache - pulsar.getLocalZkCacheService().managedLedgerListCache().clearTree(); + pulsar.getDataZkCacheService().managedLedgerListCache().clearTree(); // setup ownership to localhost doReturn(localWebServiceUrl).when(nsSvc).getWebServiceUrl(testNs, false, false, false); doReturn(true).when(nsSvc).isServiceUnitOwned(testNs); @@ -837,7 +837,7 @@ public void testSplitBundles() throws Exception { createBundledTestNamespaces(this.testProperty, this.testLocalCluster, bundledNsLocal, bundleData); final NamespaceName testNs = new NamespaceName(this.testProperty, this.testLocalCluster, bundledNsLocal); - OwnershipCache MockOwnershipCache = spy(pulsar.getNamespaceService().getOwnershipCache()); + NamespaceOwnershipCache MockOwnershipCache = spy(pulsar.getNamespaceService().getOwnershipCache()); doNothing().when(MockOwnershipCache).disableOwnership(any(NamespaceBundle.class)); Field ownership = NamespaceService.class.getDeclaredField("ownershipCache"); ownership.setAccessible(true); @@ -982,7 +982,7 @@ public void testValidateNamespaceOwnershipWithBundles() throws Exception { final NamespaceName testNs = new NamespaceName(this.testProperty, this.testLocalCluster, bundledNsLocal); mockWebUrl(localWebServiceUrl, testNs); - OwnershipCache MockOwnershipCache = spy(pulsar.getNamespaceService().getOwnershipCache()); + NamespaceOwnershipCache MockOwnershipCache = spy(pulsar.getNamespaceService().getOwnershipCache()); doNothing().when(MockOwnershipCache).disableOwnership(any(NamespaceBundle.class)); Field ownership = NamespaceService.class.getDeclaredField("ownershipCache"); ownership.setAccessible(true); @@ -1004,7 +1004,7 @@ public void testRetention() throws Exception { final NamespaceName testNs = new NamespaceName(this.testProperty, this.testLocalCluster, bundledNsLocal); mockWebUrl(localWebServiceUrl, testNs); - OwnershipCache MockOwnershipCache = spy(pulsar.getNamespaceService().getOwnershipCache()); + NamespaceOwnershipCache MockOwnershipCache = spy(pulsar.getNamespaceService().getOwnershipCache()); doNothing().when(MockOwnershipCache).disableOwnership(any(NamespaceBundle.class)); Field ownership = NamespaceService.class.getDeclaredField("ownershipCache"); ownership.setAccessible(true); @@ -1027,7 +1027,7 @@ public void testValidateDestinationOwnership() throws Exception { BundlesData bundleData = new BundlesData(Lists.newArrayList("0x00000000", "0xffffffff")); createBundledTestNamespaces(this.testProperty, this.testLocalCluster, bundledNsLocal, bundleData); final NamespaceName testNs = new NamespaceName(this.testProperty, this.testLocalCluster, bundledNsLocal); - OwnershipCache MockOwnershipCache = spy(pulsar.getNamespaceService().getOwnershipCache()); + NamespaceOwnershipCache MockOwnershipCache = spy(pulsar.getNamespaceService().getOwnershipCache()); doNothing().when(MockOwnershipCache).disableOwnership(any(NamespaceBundle.class)); Field ownership = NamespaceService.class.getDeclaredField("ownershipCache"); ownership.setAccessible(true); diff --git a/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/auth/BrokerDataAndServiceZKTest.java b/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/auth/BrokerDataAndServiceZKTest.java new file mode 100644 index 0000000000000..aebe562fa6e13 --- /dev/null +++ b/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/auth/BrokerDataAndServiceZKTest.java @@ -0,0 +1,272 @@ +/** + * Copyright 2016 Yahoo Inc. + * + * Licensed 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 com.yahoo.pulsar.broker.auth; + +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.spy; + +import java.net.URI; +import java.net.URL; +import java.util.List; +import java.util.Set; +import java.util.concurrent.TimeUnit; +import java.util.function.Supplier; + +import org.apache.bookkeeper.conf.ClientConfiguration; +import org.apache.zookeeper.MockZooKeeper; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.testng.Assert; +import org.testng.annotations.Test; +import static org.testng.Assert.fail; + +import com.google.common.collect.Lists; +import com.google.common.collect.Sets; +import com.google.common.util.concurrent.MoreExecutors; +import com.yahoo.pulsar.broker.PulsarService; +import com.yahoo.pulsar.broker.cache.LocalZooKeeperCacheService; +import com.yahoo.pulsar.broker.loadbalance.impl.SimpleLoadManagerImpl; +import com.yahoo.pulsar.broker.namespace.NamespaceService; +import com.yahoo.pulsar.client.admin.PulsarAdmin; +import com.yahoo.pulsar.client.api.Authentication; +import com.yahoo.pulsar.client.api.Consumer; +import com.yahoo.pulsar.client.api.ConsumerConfiguration; +import com.yahoo.pulsar.client.api.Message; +import com.yahoo.pulsar.client.api.Producer; +import com.yahoo.pulsar.client.api.ProducerConfiguration; +import com.yahoo.pulsar.client.api.PulsarClient; +import com.yahoo.pulsar.client.api.SubscriptionType; +import com.yahoo.pulsar.common.policies.data.ClusterData; +import com.yahoo.pulsar.common.policies.data.PropertyAdmin; +import org.apache.zookeeper.KeeperException.NoNodeException; + +public class BrokerDataAndServiceZKTest { + private static final Logger log = LoggerFactory.getLogger(BrokerDataAndServiceZKTest.class); + + /** + * Start pulsar with same data-zk and cluster-management-zk + * 1. Both zk instance should be the same in pulsar + * 2. all the znodes should be craeted correctly + * + * @throws Exception + */ + @Test + public void testPulsarWithSingleDataAndLocalZk() throws Exception { + + PulsarServiceStarterTest pulsar = new PulsarServiceStarterTest(false); + // (1) test msg produce and consume + final String property = "my-property"; + final String namespace = "my-ns"; + testProduceConsume(pulsar.pulsarClient, property, namespace); + // (2) test zk-state + Assert.assertEquals(pulsar.pulsar.getLocalZooKeeperClientFactory(), pulsar.pulsar.getZooKeeperClientFactory()); + Assert.assertEquals(pulsar.pulsar.getLocalZkCache(), pulsar.pulsar.getDataZkCache()); + List brokers = pulsar.localMockZookKeeper.getChildren(SimpleLoadManagerImpl.LOADBALANCE_BROKERS_ROOT, + null); + List nsOwnership = pulsar.localMockZookKeeper.getChildren(LocalZooKeeperCacheService.OWNER_INFO_ROOT, + null); + List localPolicies = pulsar.localMockZookKeeper + .getChildren(LocalZooKeeperCacheService.LOCAL_POLICIES_ROOT, null); + List managedLedger = pulsar.localMockZookKeeper + .getChildren(LocalZooKeeperCacheService.MANAGED_LEDGER_ROOT, null); + List counters = pulsar.localMockZookKeeper.getChildren("/counters", null); + List ledgers = pulsar.localMockZookKeeper.getChildren("/ledgers", null); + + // verify znodes are created properly + Assert.assertTrue(brokers.contains("localhost:" + pulsar.BROKER_WEBSERVICE_PORT)); + Assert.assertTrue(nsOwnership.contains(property)); + Assert.assertTrue(localPolicies.contains(property)); + Assert.assertTrue(managedLedger.contains(property)); + Assert.assertTrue(counters.contains("producer-name")); + Assert.assertFalse(ledgers.isEmpty()); + } + + /** + * Start pulsar with different data-zk and cluster-management-zk + * 1. Both zk instance should be the differemt in pulsar + * 2. all the znodes should be created correctly + * a. cluster-management-znode: /loadbalance, /namespace, /counter, /admin/local-policies + * b. data-znode: /ledgers, /managed-ledger + * + * @throws Exception + */ + @Test + public void testPulsarWithMultipleZkForDataAndLocal() throws Exception { + + PulsarServiceStarterTest pulsar = new PulsarServiceStarterTest(true); + // (1) test msg produce and consume + final String property = "my-property"; + final String namespace = "my-ns"; + testProduceConsume(pulsar.pulsarClient, property, namespace); + // (2) test zk-state + Assert.assertNotEquals(pulsar.pulsar.getLocalZooKeeperClientFactory(), + pulsar.pulsar.getZooKeeperClientFactory()); + Assert.assertNotEquals(pulsar.pulsar.getLocalZkCache(), pulsar.pulsar.getDataZkCache()); + List brokers = pulsar.localMockZookKeeper.getChildren(SimpleLoadManagerImpl.LOADBALANCE_BROKERS_ROOT, + null); + List nsOwnership = pulsar.localMockZookKeeper.getChildren(LocalZooKeeperCacheService.OWNER_INFO_ROOT, + null); + List localPolicies = pulsar.localMockZookKeeper + .getChildren(LocalZooKeeperCacheService.LOCAL_POLICIES_ROOT, null); + List managedLedger = pulsar.localMockZookKeeper + .getChildren(LocalZooKeeperCacheService.MANAGED_LEDGER_ROOT, null); + List counters = pulsar.localMockZookKeeper.getChildren("/counters", null); + List ledgers; + try { + ledgers = pulsar.localMockZookKeeper.getChildren("/ledgers", null); + fail("dataZK should not have node present"); + } catch (NoNodeException ne) { + // ok + } + + // verify cluster-management-znodes are created properly + Assert.assertTrue(brokers.contains("localhost:" + pulsar.BROKER_WEBSERVICE_PORT)); + Assert.assertTrue(nsOwnership.contains(property)); + Assert.assertTrue(localPolicies.contains(property)); + Assert.assertTrue(counters.contains("producer-name")); + Assert.assertTrue(managedLedger.isEmpty()); + + // LocalZooKeeperCacheService.initZK() => Creates : OWNER_INFO_ROOT,LOCAL_POLICIES_ROOT,MANAGED_LEDGER_ROOT (so, + // znode will be created in both zk but it will be empty) + brokers.clear(); + counters.clear(); + try { + brokers = pulsar.dataMockZookKeeper.getChildren(SimpleLoadManagerImpl.LOADBALANCE_BROKERS_ROOT, null); + counters = pulsar.dataMockZookKeeper.getChildren("/counters", null); + fail("dataZK should not have node present"); + } catch (NoNodeException ne) { + // ok + } + nsOwnership = pulsar.dataMockZookKeeper.getChildren(LocalZooKeeperCacheService.OWNER_INFO_ROOT, null); + localPolicies = pulsar.dataMockZookKeeper.getChildren(LocalZooKeeperCacheService.LOCAL_POLICIES_ROOT, null); + managedLedger = pulsar.dataMockZookKeeper.getChildren(LocalZooKeeperCacheService.MANAGED_LEDGER_ROOT, null); + ledgers = pulsar.dataMockZookKeeper.getChildren("/ledgers", null); + + // verify data-znodes are created properly + Assert.assertTrue(brokers.isEmpty()); + Assert.assertTrue(counters.isEmpty()); + Assert.assertTrue(nsOwnership.isEmpty()); + Assert.assertTrue(localPolicies.isEmpty()); + Assert.assertTrue(managedLedger.contains(property)); + Assert.assertFalse(ledgers.isEmpty()); + + } + + static class PulsarServiceStarterTest extends MockedPulsarServiceBaseTest { + + private boolean diffDataAndLocalZk = false; + + public PulsarServiceStarterTest(boolean diffDataAndLocalZk) throws Exception { + this.diffDataAndLocalZk = diffDataAndLocalZk; + setup(); + } + + @Override + protected void setup() throws Exception { + // internal-setup + initPulsar(); + com.yahoo.pulsar.client.api.ClientConfiguration clientConf = new com.yahoo.pulsar.client.api.ClientConfiguration(); + clientConf.setStatsInterval(0, TimeUnit.SECONDS); + String lookupUrl = brokerUrl.toString(); + if (isTcpLookup) { + lookupUrl = new URI("pulsar://localhost:" + BROKER_PORT).toString(); + } + pulsarClient = PulsarClient.create(lookupUrl, clientConf); + + // create test cluster, property, namespace + admin.clusters().createCluster("use", new ClusterData("http://127.0.0.1:" + BROKER_WEBSERVICE_PORT)); + admin.properties().createProperty("my-property", + new PropertyAdmin(Lists.newArrayList("appid1", "appid2"), Sets.newHashSet("use"))); + admin.namespaces().createNamespace("my-property/use/my-ns"); + } + + @Override + protected void cleanup() throws Exception { + super.internalCleanup(); + } + + protected final void initPulsar() throws Exception { + dataMockZookKeeper = createMockZooKeeper(); + if (diffDataAndLocalZk) { + localMockZookKeeper = MockZooKeeper.newInstance(MoreExecutors.sameThreadExecutor()); + } else { + localMockZookKeeper = dataMockZookKeeper; + } + mockBookKeeper = new NonClosableMockBookKeeper(new ClientConfiguration(), dataMockZookKeeper); + sameThreadOrderedSafeExecutor = new SameThreadOrderedSafeExecutor(); + startBroker(); + brokerUrl = new URL("http://" + pulsar.getAdvertisedAddress() + ":" + BROKER_WEBSERVICE_PORT); + brokerUrlTls = new URL("https://" + pulsar.getAdvertisedAddress() + ":" + BROKER_WEBSERVICE_PORT_TLS); + admin = spy(new PulsarAdmin(brokerUrl, (Authentication) null)); + } + + protected void setupBrokerMocks(PulsarService pulsar) throws Exception { + // Override default providers with mocked ones + doReturn(!diffDataAndLocalZk).when(pulsar).equalsDataAndLocalZk(); + doReturn(localMockZooKeeperClientFactory).when(pulsar).getLocalZooKeeperClientFactory(); + doReturn(mockBookKeeperClientFactory).when(pulsar).getBookKeeperClientFactory(); + Supplier namespaceServiceSupplier = () -> spy(new NamespaceService(pulsar)); + doReturn(namespaceServiceSupplier).when(pulsar).getNamespaceServiceProvider(); + doReturn(sameThreadOrderedSafeExecutor).when(pulsar).getOrderedExecutor(); + // use diff zk instance for split-zk usecase + if (diffDataAndLocalZk) { + doReturn(dataMockZooKeeperClientFactory).when(pulsar).getZooKeeperClientFactory(); + } else { + doReturn(localMockZooKeeperClientFactory).when(pulsar).getZooKeeperClientFactory(); + } + } + + } + + private void testProduceConsume(PulsarClient pulsarClient, String property, String namespace) throws Exception { + ConsumerConfiguration conf = new ConsumerConfiguration(); + conf.setSubscriptionType(SubscriptionType.Exclusive); + final String topic = String.format("persistent://%s/use/%s/my-topic1", property, namespace); + Consumer consumer = pulsarClient.subscribe(topic, "my-subscriber-name", conf); + ProducerConfiguration producerConf = new ProducerConfiguration(); + + Producer producer = pulsarClient.createProducer(topic, producerConf); + for (int i = 0; i < 10; i++) { + String message = "my-message-" + i; + producer.send(message.getBytes()); + } + + Message msg = null; + Set messageSet = Sets.newHashSet(); + for (int i = 0; i < 10; i++) { + msg = consumer.receive(5, TimeUnit.SECONDS); + String receivedMessage = new String(msg.getData()); + log.debug("Received message: [{}]", receivedMessage); + String expectedMessage = "my-message-" + i; + testMessageOrderAndDuplicates(messageSet, receivedMessage, expectedMessage); + } + // Acknowledge the consumption of all messages at once + consumer.acknowledgeCumulative(msg); + consumer.close(); + + } + + private void testMessageOrderAndDuplicates(Set messagesReceived, String receivedMessage, + String expectedMessage) { + // Make sure that messages are received in order + Assert.assertEquals(receivedMessage, expectedMessage, + "Received message " + receivedMessage + " did not match the expected message " + expectedMessage); + + // Make sure that there are no duplicates + Assert.assertTrue(messagesReceived.add(receivedMessage), "Received duplicate message " + receivedMessage); + } + +} diff --git a/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/auth/MockedPulsarServiceBaseTest.java b/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/auth/MockedPulsarServiceBaseTest.java index 87fa2621f9a7c..b26d643e453c0 100644 --- a/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/auth/MockedPulsarServiceBaseTest.java +++ b/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/auth/MockedPulsarServiceBaseTest.java @@ -55,9 +55,9 @@ public abstract class MockedPulsarServiceBaseTest { protected final ServiceConfiguration conf; - protected PulsarService pulsar; - protected PulsarAdmin admin; - protected PulsarClient pulsarClient; + public PulsarService pulsar; + public PulsarAdmin admin; + public PulsarClient pulsarClient; protected URL brokerUrl; protected URL brokerUrlTls; @@ -66,11 +66,12 @@ public abstract class MockedPulsarServiceBaseTest { protected final int BROKER_PORT = PortManager.nextFreePort(); protected final int BROKER_PORT_TLS = PortManager.nextFreePort(); - protected MockZooKeeper mockZookKeeper; - protected NonClosableMockBookKeeper mockBookKeeper; + public MockZooKeeper dataMockZookKeeper; + public MockZooKeeper localMockZookKeeper; + public NonClosableMockBookKeeper mockBookKeeper; protected boolean isTcpLookup = false; - private SameThreadOrderedSafeExecutor sameThreadOrderedSafeExecutor; + protected SameThreadOrderedSafeExecutor sameThreadOrderedSafeExecutor; public MockedPulsarServiceBaseTest() { this.conf = new ServiceConfiguration(); @@ -105,8 +106,9 @@ protected final void internalSetupForStatsTest() throws Exception { } protected final void init() throws Exception { - mockZookKeeper = createMockZooKeeper(); - mockBookKeeper = new NonClosableMockBookKeeper(new ClientConfiguration(), mockZookKeeper); + localMockZookKeeper = MockZooKeeper.newInstance(MoreExecutors.sameThreadExecutor()); + dataMockZookKeeper = createMockZooKeeper(); + mockBookKeeper = new NonClosableMockBookKeeper(new ClientConfiguration(), dataMockZookKeeper); sameThreadOrderedSafeExecutor = new SameThreadOrderedSafeExecutor(); @@ -123,7 +125,8 @@ protected final void internalCleanup() throws Exception { pulsarClient.close(); pulsar.close(); mockBookKeeper.reallyShutdow(); - mockZookKeeper.shutdown(); + dataMockZookKeeper.shutdown(); + localMockZookKeeper.shutdown(); sameThreadOrderedSafeExecutor.shutdown(); } @@ -156,7 +159,9 @@ protected PulsarService startBroker(ServiceConfiguration conf) throws Exception protected void setupBrokerMocks(PulsarService pulsar) throws Exception { // Override default providers with mocked ones - doReturn(mockZooKeeperClientFactory).when(pulsar).getZooKeeperClientFactory(); + doReturn(false).when(pulsar).equalsDataAndLocalZk(); + doReturn(localMockZooKeeperClientFactory).when(pulsar).getLocalZooKeeperClientFactory(); + doReturn(dataMockZooKeeperClientFactory).when(pulsar).getZooKeeperClientFactory(); doReturn(mockBookKeeperClientFactory).when(pulsar).getBookKeeperClientFactory(); Supplier namespaceServiceSupplier = () -> spy(new NamespaceService(pulsar)); @@ -165,7 +170,7 @@ protected void setupBrokerMocks(PulsarService pulsar) throws Exception { doReturn(sameThreadOrderedSafeExecutor).when(pulsar).getOrderedExecutor(); } - private MockZooKeeper createMockZooKeeper() throws Exception { + protected MockZooKeeper createMockZooKeeper() throws Exception { MockZooKeeper zk = MockZooKeeper.newInstance(MoreExecutors.sameThreadExecutor()); List dummyAclList = new ArrayList(0); @@ -178,7 +183,7 @@ private MockZooKeeper createMockZooKeeper() throws Exception { } // Prevent the MockBookKeeper instance from being closed when the broker is restarted within a test - private static class NonClosableMockBookKeeper extends MockBookKeeper { + protected static class NonClosableMockBookKeeper extends MockBookKeeper { public NonClosableMockBookKeeper(ClientConfiguration conf, ZooKeeper zk) throws Exception { super(conf, zk); @@ -199,17 +204,27 @@ public void reallyShutdow() { } } - protected ZooKeeperClientFactory mockZooKeeperClientFactory = new ZooKeeperClientFactory() { + protected ZooKeeperClientFactory dataMockZooKeeperClientFactory = new ZooKeeperClientFactory() { @Override public CompletableFuture create(String serverList, SessionType sessionType, int zkSessionTimeoutMillis) { // Always return the same instance (so that we don't loose the mock ZK content on broker restart - return CompletableFuture.completedFuture(mockZookKeeper); + return CompletableFuture.completedFuture(dataMockZookKeeper); } }; - private BookKeeperClientFactory mockBookKeeperClientFactory = new BookKeeperClientFactory() { + protected ZooKeeperClientFactory localMockZooKeeperClientFactory = new ZooKeeperClientFactory() { + + @Override + public CompletableFuture create(String serverList, SessionType sessionType, + int zkSessionTimeoutMillis) { + // Always return the same instance (so that we don't loose the mock ZK content on broker restart + return CompletableFuture.completedFuture(localMockZookKeeper); + } + }; + + protected BookKeeperClientFactory mockBookKeeperClientFactory = new BookKeeperClientFactory() { @Override public BookKeeper create(ServiceConfiguration conf, ZooKeeper zkClient) throws IOException { diff --git a/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/namespace/BrokerDataAndServiceZKDualOwnerShipTest.java b/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/namespace/BrokerDataAndServiceZKDualOwnerShipTest.java new file mode 100644 index 0000000000000..85720d3a88f84 --- /dev/null +++ b/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/namespace/BrokerDataAndServiceZKDualOwnerShipTest.java @@ -0,0 +1,435 @@ +/** + * Copyright 2016 Yahoo Inc. + * + * Licensed 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 com.yahoo.pulsar.broker.namespace; + +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.spy; + +import java.net.URI; +import java.net.URL; +import java.util.ArrayList; +import java.util.List; +import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import java.util.function.Supplier; + +import org.apache.bookkeeper.conf.ClientConfiguration; +import org.apache.bookkeeper.util.ZkUtils; +import org.apache.zookeeper.CreateMode; +import org.apache.zookeeper.MockZooKeeper; +import org.apache.zookeeper.data.ACL; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import static org.testng.Assert.assertTrue; +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertFalse; +import static org.testng.Assert.assertNotEquals; +import static org.testng.Assert.assertNull; +import static org.testng.Assert.assertNotNull; +import static org.testng.Assert.fail; +import org.testng.annotations.Test; + +import com.google.common.collect.Lists; +import com.google.common.collect.Sets; +import com.google.common.util.concurrent.MoreExecutors; +import com.yahoo.pulsar.broker.PulsarService; +import com.yahoo.pulsar.broker.auth.BrokerDataAndServiceZKTest; +import com.yahoo.pulsar.broker.auth.MockedPulsarServiceBaseTest; +import com.yahoo.pulsar.broker.auth.SameThreadOrderedSafeExecutor; +import com.yahoo.pulsar.broker.cache.LocalZooKeeperCacheService; +import com.yahoo.pulsar.broker.loadbalance.LoadManagerFactory; +import com.yahoo.pulsar.broker.loadbalance.impl.SimpleLoadManagerImpl; +import com.yahoo.pulsar.broker.namespace.OwnershipCacheFactory.OwnershipDualCacheFactoryImpl; +import com.yahoo.pulsar.client.admin.PulsarAdmin; +import com.yahoo.pulsar.client.api.Authentication; +import com.yahoo.pulsar.client.api.Consumer; +import com.yahoo.pulsar.client.api.ConsumerConfiguration; +import com.yahoo.pulsar.client.api.Message; +import com.yahoo.pulsar.client.api.Producer; +import com.yahoo.pulsar.client.api.ProducerConfiguration; +import com.yahoo.pulsar.client.api.PulsarClient; +import com.yahoo.pulsar.client.api.SubscriptionType; +import com.yahoo.pulsar.common.naming.DestinationName; +import com.yahoo.pulsar.common.naming.NamespaceBundle; +import com.yahoo.pulsar.common.policies.data.ClusterData; +import com.yahoo.pulsar.common.policies.data.PropertyAdmin; +import com.yahoo.pulsar.zookeeper.ZookeeperClientFactoryImpl; + +public class BrokerDataAndServiceZKDualOwnerShipTest { + private static final Logger log = LoggerFactory.getLogger(BrokerDataAndServiceZKTest.class); + + /** + * It verifies that bundle-ownership gets acquires on both the zk + * + * @throws Exception + */ + @Test + public void testOwnershipNodeCreatedInBothZK() throws Exception { + + PulsarServiceStarterTest pulsar = new PulsarServiceStarterTest(true); + // (1) test msg produce and consume + final String property = "my-property"; + final String namespace = "my-ns"; + testProduceConsume(pulsar.pulsarClient, property, namespace); + // verify both zk are different + assertNotEquals(pulsar.pulsar.getLocalZooKeeperClientFactory(), + pulsar.pulsar.getZooKeeperClientFactory()); + assertNotEquals(pulsar.pulsar.getLocalZkCache(), pulsar.pulsar.getDataZkCache()); + + // verify ownershipCache should be instance of newly injected ownershipCacheType + NamespaceOwnershipCache ownershipCache = pulsar.pulsar.getNamespaceService().getOwnershipCache(); + assertTrue(ownershipCache instanceof DualOwnershipCache); + + // verify: ownership node is created in both the cluster + List nsOwnership = null; + nsOwnership = pulsar.localMockZookKeeper.getChildren(LocalZooKeeperCacheService.OWNER_INFO_ROOT, null); + assertTrue(nsOwnership.contains(property)); + nsOwnership = pulsar.dataMockZookKeeper.getChildren(LocalZooKeeperCacheService.OWNER_INFO_ROOT, null); + assertTrue(nsOwnership.contains(property)); + } + + /** + * It makes sure that bundle-ownership should be atomic: + * + * A. ownership-znode present into secondary + * 1. Ownership node is present into secondary-zk + * 2. tryToAcquireOwnership: + * a. creates ownership-node on primary-zk + * b. Failed to create on secondary as node is present + * c. delete ownership-node on primary-zk + * 3. Verify primary-zk should not have ownership-znode + * + * B. ownership-znode present into primary + * 1. It fails to acquire ownership on secondary-zk + * @throws Exception + */ + @Test + public void testAtomicOwnershipFailOnAcquiringOnAnyZk() throws Exception { + + PulsarServiceStarterTest pulsar = new PulsarServiceStarterTest(true); + // (1) test msg produce and consume + final String property = "my-property"; + final String namespace1 = "my-ns1"; + final String namespace2 = "my-ns2"; + final String cluster = "use"; + final String topicName1 = String.format("persistent://%s/%s/%s/my-topic1", property, cluster, namespace1); + // verify both zk are different + assertNotEquals(pulsar.pulsar.getLocalZooKeeperClientFactory(), + pulsar.pulsar.getZooKeeperClientFactory()); + assertNotEquals(pulsar.pulsar.getLocalZkCache(), pulsar.pulsar.getDataZkCache()); + + /***** A. ownership-znode present into secondary ***/ + String ownershipNodePath = String.format("/namespace/%s/%s/%s/0x00000000_0xffffffff", property, cluster, + namespace1); + // create ownership on data-zk(secondary-zk) + ZkUtils.createFullPathOptimistic(pulsar.dataMockZookKeeper, ownershipNodePath, + "".getBytes(ZookeeperClientFactoryImpl.ENCODING_SCHEME), new ArrayList(0), CreateMode.PERSISTENT); + assertNotNull(pulsar.dataMockZookKeeper.getData(ownershipNodePath, null, null)); + // try to acquire ownership : it should fail as it should be atomic in sense of acquiring on both the zk + NamespaceOwnershipCache ownershipCache = pulsar.pulsar.getNamespaceService().getOwnershipCache(); + NamespaceBundle bundle = pulsar.pulsar.getNamespaceService().getBundle(DestinationName.get(topicName1)); + CompletableFuture future = ownershipCache.tryAcquiringOwnership(bundle); + // (1) Acquiring-Ownership should fail + try { + future.get(); + fail("acquire-ownership should have failed"); + } catch (Exception ne) { + if (ne.getCause() instanceof org.apache.zookeeper.KeeperException.NodeExistsException) { + // ok : node already exist on zk2 + } else { + fail("It should not fail with " + ne.getCause().getMessage()); + } + } + // (2) first zk should have deleted the ownership + List nsOwnership = null; + nsOwnership = pulsar.localMockZookKeeper.getChildren(LocalZooKeeperCacheService.OWNER_INFO_ROOT, null); + try { + pulsar.localMockZookKeeper.getData(ownershipNodePath, null, null); + fail("it should have failed as ownership node should be deleted"); + } catch (org.apache.zookeeper.KeeperException.NoNodeException nn) { + // ok + } + + /****** B. ownership-znode present into primary ******/ + ownershipNodePath = String.format("/namespace/%s/%s/%s/0x00000000_0xffffffff", property, cluster, namespace2); + // create ownership on data-zk(secondary-zk) + ZkUtils.createFullPathOptimistic(pulsar.localMockZookKeeper, ownershipNodePath, + "".getBytes(ZookeeperClientFactoryImpl.ENCODING_SCHEME), new ArrayList(0), CreateMode.PERSISTENT); + assertNotNull(pulsar.localMockZookKeeper.getData(ownershipNodePath, null, null)); + // try to acquire ownership : it should fail as it should be atomic in sense of acquiring on both the zk + final String topicName2 = String.format("persistent://%s/%s/%s/my-topic1", property, cluster, namespace2); + bundle = pulsar.pulsar.getNamespaceService().getBundle(DestinationName.get(topicName2)); + future = ownershipCache.tryAcquiringOwnership(bundle); + // (1) Acquiring-Ownership should fail + try { + future.get(); + fail("acquire-ownership should have failed"); + } catch (Exception ne) { + if (ne.getCause() instanceof org.apache.zookeeper.KeeperException.NodeExistsException) { + // ok : node already exist on zk2 + } else { + fail("It should not fail with " + ne.getCause().getMessage()); + } + } + // (2) first zk should have deleted the ownership + nsOwnership = null; + nsOwnership = pulsar.localMockZookKeeper.getChildren(LocalZooKeeperCacheService.OWNER_INFO_ROOT, null); + // ownership should fail on secondary-zk + try { + pulsar.dataMockZookKeeper.getData(ownershipNodePath, null, null); + fail("it should have failed as ownership node should be deleted"); + } catch (org.apache.zookeeper.KeeperException.NoNodeException nn) { + // ok + } + + } + + /** + * + * It verifies lookup aggregates result from local_zk and data_zk and return response. + * + * @throws Exception + */ + @Test + public void testLookupOwnership() throws Exception { + + PulsarServiceStarterTest pulsar = new PulsarServiceStarterTest(true); + // (1) test msg produce and consume + final String property = "my-property"; + final String namespace1 = "1ns"; + final String namespace2 = "2ns"; + final String cluster = "use"; + final String topicName1 = String.format("persistent://%s/%s/%s/my-topic1", property, cluster, namespace1); + + // removeOwnership on primary first then secondary : Lookup should succeed + String ownershipNodePath = String.format("/namespace/%s/%s/%s/0x00000000_0xffffffff", property, cluster, namespace1); + DualOwnershipCache ownershipCache = (DualOwnershipCache) pulsar.pulsar.getNamespaceService().getOwnershipCache(); + NamespaceBundle bundle = pulsar.pulsar.getNamespaceService().getBundle(DestinationName.get(topicName1)); + CompletableFuture future = ownershipCache.tryAcquiringOwnership(bundle); + NamespaceEphemeralData owner = future.get(); + assertNotNull(owner); + assertEquals(owner.getNativeUrl(), pulsar.pulsar.getBrokerServiceUrl()); + + // check before removing from any zk + assertNotNull(ownershipCache.getOwnedBundle(bundle)); + assertNotNull(ownershipCache.getOwnedBundles().get(ownershipNodePath)); + assertNotNull(ownershipCache.getOwnerAsync(bundle).get()); + // remove from primary zk + ownershipCache.localZkOwnershipCache.removeOwnership(bundle).get(); + assertNotNull(ownershipCache.getOwnedBundle(bundle)); + assertNotNull(ownershipCache.getOwnedBundles().get(ownershipNodePath)); + assertTrue(ownershipCache.getOwnerAsync(bundle).get().isPresent()); + // remove secondary zk + ownershipCache.dataZkOwnershipCache.removeOwnership(bundle).get(); + // it should be remove from all + assertNull(ownershipCache.getOwnedBundle(bundle)); + assertNull(ownershipCache.getOwnedBundles().get(ownershipNodePath)); + assertFalse(ownershipCache.getOwnerAsync(bundle).get().isPresent()); + + // removeOwnership on secondary first then primary: Lookup should succeed + final String topicName2 = String.format("persistent://%s/%s/%s/my-topic1", property, cluster, namespace2); + bundle = pulsar.pulsar.getNamespaceService().getBundle(DestinationName.get(topicName2)); + future = ownershipCache.tryAcquiringOwnership(bundle); + owner = future.get(); + assertNotNull(owner); + assertEquals(owner.getNativeUrl(), pulsar.pulsar.getBrokerServiceUrl()); + + ownershipNodePath = String.format("/namespace/%s/%s/%s/0x00000000_0xffffffff", property, cluster, namespace2); + assertNotNull(ownershipCache.getOwnedBundle(bundle)); + assertNotNull(ownershipCache.getOwnedBundles().get(ownershipNodePath)); + assertTrue(ownershipCache.getOwnerAsync(bundle).get().isPresent()); + ownershipCache.removeOwnership(bundle); + assertNull(ownershipCache.getOwnedBundles().get(ownershipNodePath)); + assertNull(ownershipCache.getOwnedBundle(bundle)); + assertFalse(ownershipCache.getOwnerAsync(bundle).get().isPresent()); + + } + + /** + * It verifies that LoadManager writes loadReport and register broker in both the zookeeper + * + * verify: + *

+ * 1. pulsar1-broker registered itself to only data_zk, but pulsar2-broker registered itself to both local_zk and + * data_zk + *

+ * 2. data_zk-load-balancer should return : active-brokers and load-report with both brokers, BUT + * local_zk-load-balancer should return : active-brokers and load-report with only broker1 (pulsar1) + *

+ * 3. pulsar1-broker: due to dual-write: it knows load-report and namespace-bundle of both brokers, pulsar2-broker: + * due to dual-read: it knows load-report and namespace-bundle of both brokers + * + * @throws Exception + */ + @Test + public void testLoadBalancerWriteReadOnMultipleZk() throws Exception { + + // this pulsar instance will use one ZK for cluster-management and ledger-data + PulsarServiceStarterTest pulsar1 = new PulsarServiceStarterTest(false); + // this pulsar instance will use two different zk for cluster-mngmt and ledger-data, however it will use the same data_zk as pulsar1 instance + PulsarServiceStarterTest pulsar2 = new PulsarServiceStarterTest(true, true, pulsar1.localMockZookKeeper); + // verify the data_zk instance is same for both + assertEquals(pulsar1.pulsar.getLocalZkClient(), pulsar2.pulsar.getDataZkClient()); + + Set activeBrokers1 = pulsar1.pulsar.getLoadManager().getActiveBrokers(); + // broker1-loadBalancer should give both broker in activeList as broker2 has done dual-write + assertEquals(activeBrokers1.size(), 2); + // broker2-loadBalancer should give both broker in activeList as broker2 has done dual-read + Set activeBrokers2 = pulsar2.pulsar.getLoadManager().getActiveBrokers(); + assertEquals(activeBrokers2.size(), 2); + /** + * verify: + * 1. pulsar2 has registered brokers in data_zk and local_zk + pulsar1 has just registered in local_zk + * 2. so data_zk has both brokers registered but local_zk will just have broker2 + */ + List dataZkActiveBrokers = pulsar1.localMockZookKeeper.getChildren(SimpleLoadManagerImpl.LOADBALANCE_BROKERS_ROOT, null); + List localZkActiveBrokers = pulsar2.localMockZookKeeper.getChildren(SimpleLoadManagerImpl.LOADBALANCE_BROKERS_ROOT, null); + assertEquals(dataZkActiveBrokers.size(), 2); + assertEquals(localZkActiveBrokers.size(), 1); + String pulsar1Znode = pulsar1.pulsar.getAdvertisedAddress()+":"+pulsar1.pulsar.getConfiguration().getWebServicePort(); + String pulsar2Znode = pulsar1.pulsar.getAdvertisedAddress()+":"+pulsar2.pulsar.getConfiguration().getWebServicePort(); + assertTrue(dataZkActiveBrokers.contains(pulsar1Znode)); + assertTrue(dataZkActiveBrokers.contains(pulsar2Znode)); + assertTrue(localZkActiveBrokers.contains(pulsar2Znode)); + assertFalse(localZkActiveBrokers.contains(pulsar1Znode)); + } + + + static class PulsarServiceStarterTest extends MockedPulsarServiceBaseTest { + + private boolean diffDataAndLocalZk = false; + private boolean dualLoadBalancer = false; + + public PulsarServiceStarterTest(boolean diffDataAndLocalZk) throws Exception { + this.diffDataAndLocalZk = diffDataAndLocalZk; + // initialize data-zk first + dataMockZookKeeper = createMockZooKeeper(); + setup(); + } + + public PulsarServiceStarterTest(boolean diffDataAndLocalZk, boolean dualLoadBalancer, MockZooKeeper dataMockZk) throws Exception { + this.diffDataAndLocalZk = diffDataAndLocalZk; + this.dualLoadBalancer = dualLoadBalancer; + dataMockZookKeeper = dataMockZk; + setup(); + } + + @Override + protected void setup() throws Exception { + // internal-setup + initPulsar(); + com.yahoo.pulsar.client.api.ClientConfiguration clientConf = new com.yahoo.pulsar.client.api.ClientConfiguration(); + clientConf.setStatsInterval(0, TimeUnit.SECONDS); + String lookupUrl = brokerUrl.toString(); + if (isTcpLookup) { + lookupUrl = new URI("pulsar://localhost:" + BROKER_PORT).toString(); + } + pulsarClient = PulsarClient.create(lookupUrl, clientConf); + + // create test cluster, property, namespace + try { + admin.clusters().createCluster("use", new ClusterData("http://127.0.0.1:" + BROKER_WEBSERVICE_PORT)); + admin.properties().createProperty("my-property", + new PropertyAdmin(Lists.newArrayList("appid1", "appid2"), Sets.newHashSet("use"))); + admin.namespaces().createNamespace("my-property/use/my-ns"); + } catch(Exception e){ + //ignore if cluster is already exist + } + } + + @Override + protected void cleanup() throws Exception { + super.internalCleanup(); + } + + protected final void initPulsar() throws Exception { + if (diffDataAndLocalZk) { + localMockZookKeeper = MockZooKeeper.newInstance(MoreExecutors.sameThreadExecutor()); + } else { + localMockZookKeeper = dataMockZookKeeper; + } + mockBookKeeper = new NonClosableMockBookKeeper(new ClientConfiguration(), dataMockZookKeeper); + sameThreadOrderedSafeExecutor = new SameThreadOrderedSafeExecutor(); + if(diffDataAndLocalZk) { + conf.setNamespaceOwnershipCacheFactoryProvider(OwnershipDualCacheFactoryImpl.class.getName()); + } + if(dualLoadBalancer) { + conf.setLoadBalancerFactoryProvider(LoadManagerFactory.SimpleLoadManagerDualZkImplFactory.class.getName()); + } + startBroker(); + brokerUrl = new URL("http://" + pulsar.getAdvertisedAddress() + ":" + BROKER_WEBSERVICE_PORT); + brokerUrlTls = new URL("https://" + pulsar.getAdvertisedAddress() + ":" + BROKER_WEBSERVICE_PORT_TLS); + admin = spy(new PulsarAdmin(brokerUrl, (Authentication) null)); + } + + protected void setupBrokerMocks(PulsarService pulsar) throws Exception { + // Override default providers with mocked ones + doReturn(!diffDataAndLocalZk).when(pulsar).equalsDataAndLocalZk(); + doReturn(localMockZooKeeperClientFactory).when(pulsar).getLocalZooKeeperClientFactory(); + doReturn(mockBookKeeperClientFactory).when(pulsar).getBookKeeperClientFactory(); + Supplier namespaceServiceSupplier = () -> spy(new NamespaceService(pulsar)); + doReturn(namespaceServiceSupplier).when(pulsar).getNamespaceServiceProvider(); + doReturn(sameThreadOrderedSafeExecutor).when(pulsar).getOrderedExecutor(); + // use diff zk instance for split-zk usecase + if (diffDataAndLocalZk) { + doReturn(dataMockZooKeeperClientFactory).when(pulsar).getZooKeeperClientFactory(); + } else { + doReturn(localMockZooKeeperClientFactory).when(pulsar).getZooKeeperClientFactory(); + } + } + + } + + private String testProduceConsume(PulsarClient pulsarClient, String property, String namespace) throws Exception { + ConsumerConfiguration conf = new ConsumerConfiguration(); + conf.setSubscriptionType(SubscriptionType.Exclusive); + final String topic = String.format("persistent://%s/use/%s/my-topic1", property, namespace); + Consumer consumer = pulsarClient.subscribe(topic, "my-subscriber-name", conf); + ProducerConfiguration producerConf = new ProducerConfiguration(); + + Producer producer = pulsarClient.createProducer(topic, producerConf); + for (int i = 0; i < 10; i++) { + String message = "my-message-" + i; + producer.send(message.getBytes()); + } + + Message msg = null; + Set messageSet = Sets.newHashSet(); + for (int i = 0; i < 10; i++) { + msg = consumer.receive(5, TimeUnit.SECONDS); + String receivedMessage = new String(msg.getData()); + log.debug("Received message: [{}]", receivedMessage); + String expectedMessage = "my-message-" + i; + testMessageOrderAndDuplicates(messageSet, receivedMessage, expectedMessage); + } + // Acknowledge the consumption of all messages at once + consumer.acknowledgeCumulative(msg); + consumer.close(); + return topic; + } + + private void testMessageOrderAndDuplicates(Set messagesReceived, String receivedMessage, + String expectedMessage) { + // Make sure that messages are received in order + assertEquals(receivedMessage, expectedMessage, + "Received message " + receivedMessage + " did not match the expected message " + expectedMessage); + + // Make sure that there are no duplicates + assertTrue(messagesReceived.add(receivedMessage), "Received duplicate message " + receivedMessage); + } + +} diff --git a/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/namespace/NamespaceServiceTest.java b/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/namespace/NamespaceServiceTest.java index ea4647659506f..2e4be7cb9ef14 100644 --- a/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/namespace/NamespaceServiceTest.java +++ b/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/namespace/NamespaceServiceTest.java @@ -81,7 +81,7 @@ protected void cleanup() throws Exception { @Test public void testSplitAndOwnBundles() throws Exception { - OwnershipCache MockOwnershipCache = spy(pulsar.getNamespaceService().getOwnershipCache()); + NamespaceOwnershipCache MockOwnershipCache = spy(pulsar.getNamespaceService().getOwnershipCache()); doNothing().when(MockOwnershipCache).disableOwnership(any(NamespaceBundle.class)); Field ownership = NamespaceService.class.getDeclaredField("ownershipCache"); ownership.setAccessible(true); @@ -146,7 +146,7 @@ public void testSplitAndOwnBundles() throws Exception { @Test public void testSplitMapWithRefreshedStatMap() throws Exception { - OwnershipCache MockOwnershipCache = spy(pulsar.getNamespaceService().getOwnershipCache()); + NamespaceOwnershipCache MockOwnershipCache = spy(pulsar.getNamespaceService().getOwnershipCache()); ManagedLedger ledger = mock(ManagedLedger.class); when(ledger.getCursors()).thenReturn(Lists.newArrayList()); @@ -199,7 +199,7 @@ public void testSplitMapWithRefreshedStatMap() throws Exception { @Test public void testIsServiceUnitDisabled() throws Exception { - OwnershipCache MockOwnershipCache = spy(pulsar.getNamespaceService().getOwnershipCache()); + NamespaceOwnershipCache MockOwnershipCache = spy(pulsar.getNamespaceService().getOwnershipCache()); ManagedLedger ledger = mock(ManagedLedger.class); when(ledger.getCursors()).thenReturn(Lists.newArrayList()); @@ -222,7 +222,7 @@ public void testIsServiceUnitDisabled() throws Exception { @Test public void testremoveOwnershipNamespaceBundle() throws Exception { - OwnershipCache ownershipCache = spy(pulsar.getNamespaceService().getOwnershipCache()); + NamespaceOwnershipCache ownershipCache = spy(pulsar.getNamespaceService().getOwnershipCache()); ManagedLedger ledger = mock(ManagedLedger.class); when(ledger.getCursors()).thenReturn(Lists.newArrayList()); diff --git a/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/namespace/OwnershipCacheTest.java b/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/namespace/OwnershipCacheTest.java index f146c1a6f47dd..ebd9a17b77de1 100644 --- a/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/namespace/OwnershipCacheTest.java +++ b/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/namespace/OwnershipCacheTest.java @@ -92,14 +92,14 @@ public void teardown() throws Exception { @Test public void testConstructor() { - OwnershipCache cache = new OwnershipCache(this.pulsar, bundleFactory); + NamespaceOwnershipCache cache = new OwnershipCacheImpl(this.pulsar, pulsar.getLocalZkCache(), bundleFactory); assertNotNull(cache); assertNotNull(cache.getOwnedBundles()); } @Test public void testDisableOwnership() throws Exception { - OwnershipCache cache = new OwnershipCache(this.pulsar, bundleFactory); + NamespaceOwnershipCache cache = new OwnershipCacheImpl(this.pulsar, pulsar.getLocalZkCache(), bundleFactory); NamespaceBundle testBundle = bundleFactory.getFullBundle(new NamespaceName("pulsar/test/ns-1")); assertFalse(cache.getOwnerAsync(testBundle).get().isPresent()); @@ -115,7 +115,7 @@ public void testDisableOwnership() throws Exception { @Test public void testGetOrSetOwner() throws Exception { - OwnershipCache cache = new OwnershipCache(this.pulsar, bundleFactory); + NamespaceOwnershipCache cache = new OwnershipCacheImpl(this.pulsar, pulsar.getLocalZkCache(), bundleFactory); NamespaceBundle testFullBundle = bundleFactory.getFullBundle(new NamespaceName("pulsar/test/ns-2")); // case 1: no one owns the namespace assertFalse(cache.getOwnerAsync(testFullBundle).get().isPresent()); @@ -135,7 +135,7 @@ public void testGetOrSetOwner() throws Exception { // succeeded in both cases, the ownerInfoCache will be updated (i.e. invalidated the entry) localCache.ownerInfoCache().invalidate(ServiceUnitZkUtils.path(testFullBundle)); ServiceUnitZkUtils.acquireNameSpace(zkCache.getZooKeeper(), ServiceUnitZkUtils.path(testFullBundle), - new NamespaceEphemeralData("pulsar://otherhost:8881", "pulsar://otherhost:8884", + new NamespaceEphemeralData("pulsar://otherhost:8881", "pulsar://otherhost:8884", //TODO: change to localZK "http://localhost:8080", "https://localhost:4443", false)); data1 = cache.tryAcquiringOwnership(testFullBundle).get(); assertEquals(data1.getNativeUrl(), "pulsar://otherhost:8881"); @@ -146,13 +146,13 @@ public void testGetOrSetOwner() throws Exception { @Test public void testGetOwner() throws Exception { - OwnershipCache cache = new OwnershipCache(this.pulsar, bundleFactory); + NamespaceOwnershipCache cache = new OwnershipCacheImpl(this.pulsar, pulsar.getLocalZkCache(), bundleFactory); NamespaceBundle testBundle = bundleFactory.getFullBundle(new NamespaceName("pulsar/test/ns-3")); // case 1: no one owns the namespace assertFalse(cache.getOwnerAsync(testBundle).get().isPresent()); // case 2: someone owns the namespace ServiceUnitZkUtils.acquireNameSpace(zkCache.getZooKeeper(), ServiceUnitZkUtils.path(testBundle), - new NamespaceEphemeralData("pulsar://otherhost:8881", "pulsar://otherhost:8884", + new NamespaceEphemeralData("pulsar://otherhost:8881", "pulsar://otherhost:8884", //TODO: change to localZK "http://otherhost:8080", "https://otherhost:4443", false)); // try to acquire, which will load the read-only cache @@ -174,7 +174,7 @@ public void testGetOwner() throws Exception { @Test public void testGetOwnedServiceUnit() throws Exception { - OwnershipCache cache = new OwnershipCache(this.pulsar, bundleFactory); + NamespaceOwnershipCache cache = new OwnershipCacheImpl(this.pulsar, pulsar.getLocalZkCache(), bundleFactory); NamespaceName testNs = new NamespaceName("pulsar/test/ns-5"); NamespaceBundle testBundle = bundleFactory.getFullBundle(testNs); // case 1: no one owns the namespace @@ -188,7 +188,7 @@ public void testGetOwnedServiceUnit() throws Exception { } // case 2: someone else owns the namespace ServiceUnitZkUtils.acquireNameSpace(zkCache.getZooKeeper(), ServiceUnitZkUtils.path(testBundle), - new NamespaceEphemeralData("pulsar://otherhost:8881", "pulsar://otherhost:8884", + new NamespaceEphemeralData("pulsar://otherhost:8881", "pulsar://otherhost:8884", //TODO: change to localZK "http://otherhost:8080", "https://otherhost:4443", false)); try { checkNotNull(cache.getOwnedBundle(testBundle)); @@ -220,7 +220,7 @@ public void testGetOwnedServiceUnit() throws Exception { @Test public void testGetOwnedServiceUnits() throws Exception { - OwnershipCache cache = new OwnershipCache(this.pulsar, bundleFactory); + NamespaceOwnershipCache cache = new OwnershipCacheImpl(this.pulsar, pulsar.getLocalZkCache(), bundleFactory); NamespaceName testNs = new NamespaceName("pulsar/test/ns-6"); NamespaceBundle testBundle = bundleFactory.getFullBundle(testNs); // case 1: no one owns the namespace @@ -230,7 +230,7 @@ public void testGetOwnedServiceUnits() throws Exception { // case 2: someone else owns the namespace ServiceUnitZkUtils.acquireNameSpace(zkCache.getZooKeeper(), ServiceUnitZkUtils.path(testBundle), - new NamespaceEphemeralData("pulsar://otherhost:8881", "pulsar://otherhost:8884", + new NamespaceEphemeralData("pulsar://otherhost:8881", "pulsar://otherhost:8884", //TODO: change to localZK "http://otherhost:8080", "https://otherhost:4443", false)); assertTrue(cache.getOwnedBundles().isEmpty()); // try to acquire, which will load the read-only cache @@ -252,7 +252,7 @@ public void testGetOwnedServiceUnits() throws Exception { @Test public void testRemoveOwnership() throws Exception { - OwnershipCache cache = new OwnershipCache(this.pulsar, bundleFactory); + NamespaceOwnershipCache cache = new OwnershipCacheImpl(this.pulsar, pulsar.getLocalZkCache(), bundleFactory); NamespaceName testNs = new NamespaceName("pulsar/test/ns-7"); NamespaceBundle bundle = bundleFactory.getFullBundle(testNs); // case 1: no one owns the namespace diff --git a/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/service/PersistentDispatcherFailoverConsumerTest.java b/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/service/PersistentDispatcherFailoverConsumerTest.java index d8576be9d78bc..19584766e7643 100644 --- a/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/service/PersistentDispatcherFailoverConsumerTest.java +++ b/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/service/PersistentDispatcherFailoverConsumerTest.java @@ -87,7 +87,7 @@ public void setup() throws Exception { doReturn(mlFactoryMock).when(pulsar).getManagedLedgerFactory(); ZooKeeper mockZk = mock(ZooKeeper.class); - doReturn(mockZk).when(pulsar).getZkClient(); + doReturn(mockZk).when(pulsar).getDataZkClient(); configCacheService = mock(ConfigurationCacheService.class); @SuppressWarnings("unchecked") diff --git a/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/service/PersistentTopicConcurrentTest.java b/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/service/PersistentTopicConcurrentTest.java index 32497ceb2a111..3d059116b6722 100644 --- a/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/service/PersistentTopicConcurrentTest.java +++ b/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/service/PersistentTopicConcurrentTest.java @@ -85,7 +85,7 @@ public void setup(Method m) throws Exception { doReturn(mlFactoryMock).when(pulsar).getManagedLedgerFactory(); ZooKeeper mockZk = mock(ZooKeeper.class); - doReturn(mockZk).when(pulsar).getZkClient(); + doReturn(mockZk).when(pulsar).getDataZkClient(); brokerService = spy(new BrokerService(pulsar)); doReturn(brokerService).when(pulsar).getBrokerService(); diff --git a/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/service/PersistentTopicTest.java b/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/service/PersistentTopicTest.java index d3e7517af32de..81d6de1896e99 100644 --- a/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/service/PersistentTopicTest.java +++ b/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/service/PersistentTopicTest.java @@ -116,7 +116,7 @@ public void setup() throws Exception { doReturn(mlFactoryMock).when(pulsar).getManagedLedgerFactory(); ZooKeeper mockZk = mock(ZooKeeper.class); - doReturn(mockZk).when(pulsar).getZkClient(); + doReturn(mockZk).when(pulsar).getDataZkClient(); configCacheService = mock(ConfigurationCacheService.class); @SuppressWarnings("unchecked") diff --git a/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/service/ReplicatorTest.java b/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/service/ReplicatorTest.java index 2749ff61ee1b6..f1de15a33ff9d 100644 --- a/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/service/ReplicatorTest.java +++ b/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/service/ReplicatorTest.java @@ -48,7 +48,7 @@ import com.yahoo.pulsar.broker.namespace.NamespaceService; import com.yahoo.pulsar.broker.namespace.OwnedBundle; -import com.yahoo.pulsar.broker.namespace.OwnershipCache; +import com.yahoo.pulsar.broker.namespace.NamespaceOwnershipCache; import com.yahoo.pulsar.broker.service.persistent.PersistentReplicator; import com.yahoo.pulsar.broker.service.persistent.PersistentTopic; import com.yahoo.pulsar.client.admin.PulsarAdminException.PreconditionFailedException; @@ -214,7 +214,7 @@ public void testConfigChangeNegativeCases() throws Exception { // Set up field access to internal namespace state in NamespaceService Field ownershipCacheField = NamespaceService.class.getDeclaredField("ownershipCache"); ownershipCacheField.setAccessible(true); - OwnershipCache ownerCache = (OwnershipCache) ownershipCacheField.get(pulsar1.getNamespaceService()); + NamespaceOwnershipCache ownerCache = (NamespaceOwnershipCache) ownershipCacheField.get(pulsar1.getNamespaceService()); Assert.assertNotNull(pulsar1, "pulsar1 is null"); Assert.assertNotNull(pulsar1.getNamespaceService(), "pulsar1.getNamespaceService() is null"); NamespaceBundle globalNsBundle = pulsar1.getNamespaceService().getNamespaceBundleFactory() diff --git a/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/service/ServerCnxTest.java b/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/service/ServerCnxTest.java index 345b3ca9c4bd8..44f2376fde0a5 100644 --- a/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/service/ServerCnxTest.java +++ b/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/service/ServerCnxTest.java @@ -135,7 +135,7 @@ public void setup() throws Exception { doReturn(mlFactoryMock).when(pulsar).getManagedLedgerFactory(); ZooKeeper mockZk = mock(ZooKeeper.class); - doReturn(mockZk).when(pulsar).getZkClient(); + doReturn(mockZk).when(pulsar).getDataZkClient(); configCacheService = mock(ConfigurationCacheService.class); @SuppressWarnings("unchecked") diff --git a/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/web/WebServiceTest.java b/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/web/WebServiceTest.java index 948fa14734ca8..9012985e6bca5 100644 --- a/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/web/WebServiceTest.java +++ b/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/web/WebServiceTest.java @@ -292,13 +292,14 @@ private void setupEnv(boolean enableFilter, String minApiVersion, boolean allowU config.setAdvertisedAddress("localhost"); // TLS certificate expects localhost pulsar = spy(new PulsarService(config)); doReturn(new MockedZooKeeperClientFactoryImpl()).when(pulsar).getZooKeeperClientFactory(); + doReturn(true).when(pulsar).equalsDataAndLocalZk(); pulsar.start(); try { - pulsar.getZkClient().delete("/minApiVersion", -1); + pulsar.getLocalZkClient().delete("/minApiVersion", -1); } catch (Exception ex) { } - pulsar.getZkClient().create("/minApiVersion", minApiVersion.getBytes(), null, CreateMode.PERSISTENT); + pulsar.getDataZkClient().create("/minApiVersion", minApiVersion.getBytes(), null, CreateMode.PERSISTENT); String serviceUrl = BROKER_URL_BASE; ClientConfiguration clientConfig = new ClientConfiguration(); diff --git a/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/zookeeper/ZooKeeperSessionExpireRecoveryTest.java b/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/zookeeper/ZooKeeperSessionExpireRecoveryTest.java index 98abfe4a2131c..ca7ab554a3cce 100644 --- a/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/zookeeper/ZooKeeperSessionExpireRecoveryTest.java +++ b/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/zookeeper/ZooKeeperSessionExpireRecoveryTest.java @@ -51,7 +51,7 @@ public void testSessionExpired() throws Exception { assertEquals(admin.clusters().getClusters(), Lists.newArrayList("my-cluster")); - mockZookKeeper.failNow(Code.SESSIONEXPIRED); + dataMockZookKeeper.failNow(Code.SESSIONEXPIRED); assertEquals(admin.clusters().getClusters(), Lists.newArrayList("my-cluster")); diff --git a/pulsar-broker/src/test/java/com/yahoo/pulsar/client/api/BrokerServiceLookupTest.java b/pulsar-broker/src/test/java/com/yahoo/pulsar/client/api/BrokerServiceLookupTest.java index 4a22a6470cf8b..d2c315b36c66d 100644 --- a/pulsar-broker/src/test/java/com/yahoo/pulsar/client/api/BrokerServiceLookupTest.java +++ b/pulsar-broker/src/test/java/com/yahoo/pulsar/client/api/BrokerServiceLookupTest.java @@ -30,6 +30,7 @@ import java.security.PrivateKey; import java.security.SecureRandom; import java.security.cert.Certificate; +import java.util.ArrayList; import java.util.HashMap; import java.util.Map; import java.util.Set; @@ -43,6 +44,9 @@ import javax.net.ssl.TrustManager; import org.apache.bookkeeper.test.PortManager; +import org.apache.bookkeeper.util.ZkUtils; +import org.apache.zookeeper.CreateMode; +import org.apache.zookeeper.data.ACL; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.testng.Assert; @@ -68,6 +72,7 @@ import com.yahoo.pulsar.common.util.SecurityUtility; import com.yahoo.pulsar.discovery.service.DiscoveryService; import com.yahoo.pulsar.discovery.service.server.ServiceConfig; +import com.yahoo.pulsar.zookeeper.ZookeeperClientFactoryImpl; import io.netty.handler.ssl.util.InsecureTrustManagerFactory; @@ -447,7 +452,7 @@ public void testDiscoveryLookup() throws Exception { config.setServicePort(nextFreePort()); config.setBindOnLocalhost(true); DiscoveryService discoveryService = spy(new DiscoveryService(config)); - doReturn(mockZooKeeperClientFactory).when(discoveryService).getZooKeeperClientFactory(); + doReturn(localMockZooKeeperClientFactory).when(discoveryService).getZooKeeperClientFactory(); discoveryService.start(); // (2) lookup using discovery service @@ -510,7 +515,7 @@ public void testDiscoveryLookupTls() throws Exception { config.setTlsCertificateFilePath(TLS_SERVER_CERT_FILE_PATH); config.setTlsKeyFilePath(TLS_SERVER_KEY_FILE_PATH); DiscoveryService discoveryService = spy(new DiscoveryService(config)); - doReturn(mockZooKeeperClientFactory).when(discoveryService).getZooKeeperClientFactory(); + doReturn(localMockZooKeeperClientFactory).when(discoveryService).getZooKeeperClientFactory(); discoveryService.start(); // (3) lookup using discovery service @@ -567,7 +572,19 @@ public void testDiscoveryLookupAuthAndAuthSuccess() throws Exception { config.setAuthenticationEnabled(true); config.setAuthorizationEnabled(true); DiscoveryService discoveryService = spy(new DiscoveryService(config)); - doReturn(mockZooKeeperClientFactory).when(discoveryService).getZooKeeperClientFactory(); + + // as discoveryService uses same ZookeeperClientFactory for global and local ZK + // combine both zk as one for testing : and then create cluster/property on localZK + dataMockZooKeeperClientFactory = localMockZooKeeperClientFactory; + ZkUtils.createFullPathOptimistic(localMockZookKeeper, "/admin/clusters", + "".getBytes(ZookeeperClientFactoryImpl.ENCODING_SCHEME), new ArrayList(0), CreateMode.PERSISTENT); + ZkUtils.createFullPathOptimistic(localMockZookKeeper, "/admin/policies", + "".getBytes(ZookeeperClientFactoryImpl.ENCODING_SCHEME), new ArrayList(0), CreateMode.PERSISTENT); + doReturn(pulsar.getLocalZkCache()).when(pulsar).getGlobalZkCache(); + admin.clusters().createCluster("use", new ClusterData("http://127.0.0.1:" + BROKER_WEBSERVICE_PORT)); + admin.properties().createProperty("my-property1", new PropertyAdmin(Lists.newArrayList("appid1", "appid2"), Sets.newHashSet("use2"))); + + doReturn(localMockZooKeeperClientFactory).when(discoveryService).getZooKeeperClientFactory(); discoveryService.start(); // (2) lookup using discovery service @@ -596,9 +613,9 @@ public void start() throws PulsarClientException { }); PulsarClient pulsarClient = PulsarClient.create(discoverySvcUrl, clientConfig); - Consumer consumer = pulsarClient.subscribe("persistent://my-property/use2/my-ns/my-topic1", + Consumer consumer = pulsarClient.subscribe("persistent://my-property1/use2/my-ns/my-topic1", "my-subscriber-name", new ConsumerConfiguration()); - Producer producer = pulsarClient.createProducer("persistent://my-property/use2/my-ns/my-topic1", + Producer producer = pulsarClient.createProducer("persistent://my-property1/use2/my-ns/my-topic1", new ProducerConfiguration()); for (int i = 0; i < 10; i++) { String message = "my-message-" + i; @@ -633,7 +650,7 @@ public void testDiscoveryLookupAuthenticationFailure() throws Exception { config.setAuthenticationEnabled(true); config.setAuthorizationEnabled(true); DiscoveryService discoveryService = spy(new DiscoveryService(config)); - doReturn(mockZooKeeperClientFactory).when(discoveryService).getZooKeeperClientFactory(); + doReturn(localMockZooKeeperClientFactory).when(discoveryService).getZooKeeperClientFactory(); discoveryService.start(); // (2) lookup using discovery service final String discoverySvcUrl = discoveryService.getServiceUrl(); @@ -683,7 +700,7 @@ public void testDiscoveryLookupAuthorizationFailure() throws Exception { config.setAuthenticationEnabled(true); config.setAuthorizationEnabled(true); DiscoveryService discoveryService = spy(new DiscoveryService(config)); - doReturn(mockZooKeeperClientFactory).when(discoveryService).getZooKeeperClientFactory(); + doReturn(localMockZooKeeperClientFactory).when(discoveryService).getZooKeeperClientFactory(); discoveryService.start(); // (2) lookup using discovery service final String discoverySvcUrl = discoveryService.getServiceUrl(); diff --git a/pulsar-broker/src/test/java/com/yahoo/pulsar/client/impl/BrokerClientIntegrationTest.java b/pulsar-broker/src/test/java/com/yahoo/pulsar/client/impl/BrokerClientIntegrationTest.java index a466b074a0370..71e47a73a0d08 100644 --- a/pulsar-broker/src/test/java/com/yahoo/pulsar/client/impl/BrokerClientIntegrationTest.java +++ b/pulsar-broker/src/test/java/com/yahoo/pulsar/client/impl/BrokerClientIntegrationTest.java @@ -30,7 +30,7 @@ import org.testng.annotations.BeforeMethod; import org.testng.annotations.Test; -import com.yahoo.pulsar.broker.namespace.OwnershipCache; +import com.yahoo.pulsar.broker.namespace.NamespaceOwnershipCache; import com.yahoo.pulsar.client.api.ConsumerConfiguration; import com.yahoo.pulsar.client.api.ProducerConfiguration; import com.yahoo.pulsar.client.api.ProducerConsumerBase; @@ -180,7 +180,7 @@ public void testCloseBrokerService() throws Exception { pulsar.getBrokerService().close(); // [1] OwnershipCache should not contain any more namespaces - OwnershipCache ownershipCache = pulsar.getNamespaceService().getOwnershipCache(); + NamespaceOwnershipCache ownershipCache = pulsar.getNamespaceService().getOwnershipCache(); assertTrue(ownershipCache.getOwnedBundles().keySet().isEmpty()); // [2] All clients must be disconnected and in connecting state diff --git a/pulsar-broker/src/test/java/com/yahoo/pulsar/discovery/service/web/DiscoveryServiceWebTest.java b/pulsar-broker/src/test/java/com/yahoo/pulsar/discovery/service/web/DiscoveryServiceWebTest.java index 2fe7c7a15298f..b2da967b0bb59 100644 --- a/pulsar-broker/src/test/java/com/yahoo/pulsar/discovery/service/web/DiscoveryServiceWebTest.java +++ b/pulsar-broker/src/test/java/com/yahoo/pulsar/discovery/service/web/DiscoveryServiceWebTest.java @@ -80,7 +80,7 @@ public void testRiderectUrlWithServerStarted() throws Exception { ServiceConfig config = new ServiceConfig(); config.setWebServicePort(port); ServerManager server = new ServerManager(config); - DiscoveryZooKeeperClientFactoryImpl.zk = mockZookKeeper; + DiscoveryZooKeeperClientFactoryImpl.zk = localMockZookKeeper; Map params = new TreeMap<>(); params.put("zookeeperServers", ""); params.put("zookeeperClientFactoryClass", DiscoveryZooKeeperClientFactoryImpl.class.getName()); diff --git a/pulsar-broker/src/test/java/com/yahoo/pulsar/websocket/proxy/ProxyAuthenticationTest.java b/pulsar-broker/src/test/java/com/yahoo/pulsar/websocket/proxy/ProxyAuthenticationTest.java index 18c5afcbb0340..8e5464f50583e 100644 --- a/pulsar-broker/src/test/java/com/yahoo/pulsar/websocket/proxy/ProxyAuthenticationTest.java +++ b/pulsar-broker/src/test/java/com/yahoo/pulsar/websocket/proxy/ProxyAuthenticationTest.java @@ -64,7 +64,7 @@ public void setup() throws Exception { config.setAuthenticationProviders( Sets.newHashSet("com.yahoo.pulsar.websocket.proxy.MockAuthenticationProvider")); service = spy(new WebSocketService(config)); - doReturn(mockZooKeeperClientFactory).when(service).getZooKeeperClientFactory(); + doReturn(dataMockZooKeeperClientFactory).when(service).getZooKeeperClientFactory(); proxyServer = new ProxyServer(config); WebSocketServiceStarter.start(proxyServer, service); log.info("Proxy Server Started"); diff --git a/pulsar-broker/src/test/java/com/yahoo/pulsar/websocket/proxy/ProxyAuthorizationTest.java b/pulsar-broker/src/test/java/com/yahoo/pulsar/websocket/proxy/ProxyAuthorizationTest.java index 14057c88e9bcc..017d385fc4763 100644 --- a/pulsar-broker/src/test/java/com/yahoo/pulsar/websocket/proxy/ProxyAuthorizationTest.java +++ b/pulsar-broker/src/test/java/com/yahoo/pulsar/websocket/proxy/ProxyAuthorizationTest.java @@ -61,7 +61,7 @@ protected void setup() throws Exception { config.setClusterName("c1"); config.setWebServicePort(TEST_PORT); service = spy(new WebSocketService(config)); - doReturn(mockZooKeeperClientFactory).when(service).getZooKeeperClientFactory(); + doReturn(dataMockZooKeeperClientFactory).when(service).getZooKeeperClientFactory(); service.start(); } diff --git a/pulsar-broker/src/test/java/com/yahoo/pulsar/websocket/proxy/ProxyPublishConsumeTest.java b/pulsar-broker/src/test/java/com/yahoo/pulsar/websocket/proxy/ProxyPublishConsumeTest.java index 1a5d74f887b24..7027dcdc6dd83 100644 --- a/pulsar-broker/src/test/java/com/yahoo/pulsar/websocket/proxy/ProxyPublishConsumeTest.java +++ b/pulsar-broker/src/test/java/com/yahoo/pulsar/websocket/proxy/ProxyPublishConsumeTest.java @@ -56,7 +56,7 @@ public void setup() throws Exception { config.setClusterName("use"); config.setGlobalZookeeperServers("dummy-zk-servers"); service = spy(new WebSocketService(config)); - doReturn(mockZooKeeperClientFactory).when(service).getZooKeeperClientFactory(); + doReturn(dataMockZooKeeperClientFactory).when(service).getZooKeeperClientFactory(); proxyServer = new ProxyServer(config); WebSocketServiceStarter.start(proxyServer, service); log.info("Proxy Server Started"); diff --git a/pulsar-broker/src/test/java/com/yahoo/pulsar/websocket/proxy/ProxyPublishConsumeTls.java b/pulsar-broker/src/test/java/com/yahoo/pulsar/websocket/proxy/ProxyPublishConsumeTls.java index caf40ac894e59..da6071fd20ad8 100644 --- a/pulsar-broker/src/test/java/com/yahoo/pulsar/websocket/proxy/ProxyPublishConsumeTls.java +++ b/pulsar-broker/src/test/java/com/yahoo/pulsar/websocket/proxy/ProxyPublishConsumeTls.java @@ -73,7 +73,7 @@ public void setup() throws Exception { config.setClusterName("use"); config.setGlobalZookeeperServers("dummy-zk-servers"); service = spy(new WebSocketService(config)); - doReturn(mockZooKeeperClientFactory).when(service).getZooKeeperClientFactory(); + doReturn(dataMockZooKeeperClientFactory).when(service).getZooKeeperClientFactory(); proxyServer = new ProxyServer(config); WebSocketServiceStarter.start(proxyServer, service); log.info("Proxy Server Started"); diff --git a/pulsar-broker/src/test/java/com/yahoo/pulsar/websocket/proxy/ProxyPublishConsumeWithoutZKTest.java b/pulsar-broker/src/test/java/com/yahoo/pulsar/websocket/proxy/ProxyPublishConsumeWithoutZKTest.java index 8f7e6f0c8ebf7..4284647937efb 100644 --- a/pulsar-broker/src/test/java/com/yahoo/pulsar/websocket/proxy/ProxyPublishConsumeWithoutZKTest.java +++ b/pulsar-broker/src/test/java/com/yahoo/pulsar/websocket/proxy/ProxyPublishConsumeWithoutZKTest.java @@ -57,7 +57,7 @@ public void setup() throws Exception { config.setServiceUrl(pulsar.getWebServiceAddress()); config.setServiceUrlTls(pulsar.getWebServiceAddressTls()); service = spy(new WebSocketService(config)); - doReturn(mockZooKeeperClientFactory).when(service).getZooKeeperClientFactory(); + doReturn(dataMockZooKeeperClientFactory).when(service).getZooKeeperClientFactory(); proxyServer = new ProxyServer(config); WebSocketServiceStarter.start(proxyServer, service); log.info("Proxy Server Started"); diff --git a/pulsar-zookeeper-utils/src/test/java/com/yahoo/pulsar/zookeeper/LocalZooKeeperConnectionServiceTest.java b/pulsar-zookeeper-utils/src/test/java/com/yahoo/pulsar/zookeeper/LocalZooKeeperConnectionServiceTest.java index 58a84ae581542..1d22ae74dfc43 100644 --- a/pulsar-zookeeper-utils/src/test/java/com/yahoo/pulsar/zookeeper/LocalZooKeeperConnectionServiceTest.java +++ b/pulsar-zookeeper-utils/src/test/java/com/yahoo/pulsar/zookeeper/LocalZooKeeperConnectionServiceTest.java @@ -47,7 +47,7 @@ void testSimpleZooKeeperConnection() throws Exception { localZkConnectionService.start(null); // Get ZooKeeper client - MockZooKeeper zk = (MockZooKeeper) localZkConnectionService.getLocalZooKeeper(); + MockZooKeeper zk = (MockZooKeeper) localZkConnectionService.getLocalZooKeeper(); //TODO: change to local zk // Check status assertTrue(zk.getState().isConnected());