diff --git a/pom.xml b/pom.xml index 13a2bd97c29ae..85119beb6ef29 100644 --- a/pom.xml +++ b/pom.xml @@ -479,6 +479,12 @@ flexible messaging model and an intuitive client API. aspectjweaver ${aspectj.version} + + + com.ea.agentloader + ea-agent-loader + 1.0.2 + diff --git a/pulsar-broker/pom.xml b/pulsar-broker/pom.xml index c0d907bc079c5..315dfdddffbc8 100644 --- a/pulsar-broker/pom.xml +++ b/pulsar-broker/pom.xml @@ -215,6 +215,22 @@ java-semver + + + org.aspectj + aspectjrt + + + + org.aspectj + aspectjweaver + + + + com.ea.agentloader + ea-agent-loader + + org.mockito mockito-core @@ -236,6 +252,31 @@ + + org.codehaus.mojo + aspectj-maven-plugin + 1.10 + + 1.8 + 1.8 + 1.8 + true + + + org.apache.zookeeper + zookeeper + + + + + + process-sources + + compile + + + + org.apache.maven.plugins maven-surefire-plugin @@ -283,6 +324,35 @@ + + + + + org.eclipse.m2e + lifecycle-mapping + 1.0.0 + + + + + + org.codehaus.mojo + aspectj-maven-plugin + [1.10,) + + compile + + + + + + + + + + + + src/main/resources diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/PulsarBrokerStarter.java b/pulsar-broker/src/main/java/org/apache/pulsar/PulsarBrokerStarter.java index d66433266b086..78a0b29c5562f 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/PulsarBrokerStarter.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/PulsarBrokerStarter.java @@ -26,10 +26,13 @@ import org.apache.pulsar.broker.PulsarServerException; import org.apache.pulsar.broker.PulsarService; import org.apache.pulsar.broker.ServiceConfiguration; +import org.aspectj.weaver.loadtime.Agent; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.slf4j.bridge.SLF4JBridgeHandler; +import com.ea.agentloader.AgentLoader; + public class PulsarBrokerStarter { private static ServiceConfiguration loadConfig(String configFile) throws Exception { @@ -53,6 +56,9 @@ public static void main(String[] args) throws Exception { String configFile = args[0]; ServiceConfiguration config = loadConfig(configFile); + // load aspectj-weaver agent for instrumentation + AgentLoader.loadAgentClass(Agent.class.getName(), null); + @SuppressWarnings("resource") final PulsarService service = new PulsarService(config); Runtime.getRuntime().addShutdownHook(service.getShutdownService()); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/PulsarStandaloneStarter.java b/pulsar-broker/src/main/java/org/apache/pulsar/PulsarStandaloneStarter.java index c65415e150a8f..ef934a7c7e33e 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/PulsarStandaloneStarter.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/PulsarStandaloneStarter.java @@ -21,7 +21,6 @@ import static org.apache.commons.lang3.StringUtils.isBlank; import java.io.FileInputStream; -import java.net.URI; import java.net.URL; import org.apache.pulsar.broker.PulsarService; @@ -33,11 +32,13 @@ import org.apache.pulsar.common.policies.data.ClusterData; import org.apache.pulsar.common.policies.data.PropertyAdmin; import org.apache.pulsar.zookeeper.LocalBookkeeperEnsemble; +import org.aspectj.weaver.loadtime.Agent; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import com.beust.jcommander.JCommander; import com.beust.jcommander.Parameter; +import com.ea.agentloader.AgentLoader; import com.google.common.collect.Lists; import com.google.common.collect.Sets; @@ -155,6 +156,9 @@ void start() throws Exception { return; } + // load aspectj-weaver agent for instrumentation + AgentLoader.loadAgentClass(Agent.class.getName(), null); + // Start Broker broker = new PulsarService(config); broker.start(); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java index fbd81a2a614b4..89b0f7b9651d7 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java @@ -33,18 +33,15 @@ import org.apache.bookkeeper.mledger.ManagedLedgerFactory; import org.apache.bookkeeper.util.OrderedSafeExecutor; -import org.apache.pulsar.broker.PulsarServerException; -import org.apache.pulsar.broker.ServiceConfiguration; -import org.apache.pulsar.broker.ServiceConfigurationUtils; import org.apache.pulsar.broker.admin.AdminResource; import org.apache.pulsar.broker.cache.ConfigurationCacheService; import org.apache.pulsar.broker.cache.LocalZooKeeperCacheService; import org.apache.pulsar.broker.loadbalance.LeaderElectionService; +import org.apache.pulsar.broker.loadbalance.LeaderElectionService.LeaderListener; import org.apache.pulsar.broker.loadbalance.LoadManager; import org.apache.pulsar.broker.loadbalance.LoadReportUpdaterTask; import org.apache.pulsar.broker.loadbalance.LoadResourceQuotaUpdaterTask; import org.apache.pulsar.broker.loadbalance.LoadSheddingTask; -import org.apache.pulsar.broker.loadbalance.LeaderElectionService.LeaderListener; import org.apache.pulsar.broker.loadbalance.impl.SimpleLoadManagerImpl; import org.apache.pulsar.broker.namespace.NamespaceService; import org.apache.pulsar.broker.service.BrokerService; @@ -67,7 +64,7 @@ import org.apache.pulsar.zookeeper.LocalZooKeeperConnectionService; import org.apache.pulsar.zookeeper.ZooKeeperCache; import org.apache.pulsar.zookeeper.ZooKeeperClientFactory; -import org.apache.pulsar.zookeeper.ZookeeperClientFactoryImpl; +import org.apache.pulsar.zookeeper.ZookeeperBkClientFactoryImpl; import org.apache.zookeeper.ZooKeeper; import org.eclipse.jetty.servlet.ServletHolder; import org.slf4j.Logger; @@ -564,7 +561,7 @@ public LocalZooKeeperCacheService getLocalZkCacheService() { public ZooKeeperClientFactory getZooKeeperClientFactory() { if (zkClientFactory == null) { - zkClientFactory = new ZookeeperClientFactoryImpl(); + zkClientFactory = new ZookeeperBkClientFactoryImpl(); } // Return default factory return zkClientFactory; diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java index 15b7120786fd0..0cbab32410079 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java @@ -71,6 +71,7 @@ import org.apache.pulsar.broker.service.persistent.PersistentTopic; import org.apache.pulsar.broker.stats.ClusterReplicationMetrics; import org.apache.pulsar.broker.web.PulsarWebResource; +import org.apache.pulsar.broker.zookeeper.aspectj.ClientCnxnAspect; import org.apache.pulsar.client.api.ClientConfiguration; import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.PulsarClientException; @@ -189,6 +190,10 @@ public BrokerService(PulsarService pulsar) throws Exception { this.multiLayerTopicsMap = new ConcurrentOpenHashMap<>(); this.pulsarStats = new PulsarStats(pulsar); + // register listener to capture zk-latency + ClientCnxnAspect.addListener((eventType, latencyMs) -> { + this.pulsarStats.recordZkLatencyTimeValue(eventType, latencyMs); + }); this.offlineTopicStatCache = new ConcurrentOpenHashMap<>(); final DefaultThreadFactory acceptorThreadFactory = new DefaultThreadFactory("pulsar-acceptor"); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarStats.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarStats.java index b3e60f6e49192..9b713f67338b1 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarStats.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarStats.java @@ -29,6 +29,7 @@ import org.apache.pulsar.broker.stats.BrokerOperabilityMetrics; import org.apache.pulsar.broker.stats.ClusterReplicationMetrics; import org.apache.pulsar.broker.stats.NamespaceStats; +import org.apache.pulsar.broker.zookeeper.aspectj.ClientCnxnAspect.EventType; import org.apache.pulsar.common.naming.NamespaceBundle; import org.apache.pulsar.common.stats.Metrics; import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap; @@ -190,4 +191,16 @@ public void recordTopicLoadTimeValue(String topic, long topicLoadLatencyMs) { log.warn("Exception while recording topic load time for topic {}, {}", topic, ex.getMessage()); } } + + public void recordZkLatencyTimeValue(EventType eventType, long latencyMs) { + try { + if (EventType.write.equals(eventType)) { + brokerOperabilityMetrics.recordZkWriteLatencyTimeValue(latencyMs); + } else if (EventType.read.equals(eventType)) { + brokerOperabilityMetrics.recordZkReadLatencyTimeValue(latencyMs); + } + } catch (Exception ex) { + log.warn("Exception while recording zk-latency {}, {}", eventType, ex.getMessage()); + } + } } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/BrokerOperabilityMetrics.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/BrokerOperabilityMetrics.java index 8f2809c94efee..de988b7d858ac 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/BrokerOperabilityMetrics.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/BrokerOperabilityMetrics.java @@ -23,6 +23,7 @@ import java.util.ArrayList; import java.util.List; import java.util.Map; +import java.util.concurrent.TimeUnit; import org.apache.pulsar.common.stats.Metrics; @@ -31,13 +32,17 @@ public class BrokerOperabilityMetrics { private final List metricsList; private final String localCluster; - private final TopicLoadStats topicLoadStats; + private final DimensionStats topicLoadStats; + private final DimensionStats zkWriteLatencyStats; + private final DimensionStats zkReadLatencyStats; private final String brokerName; public BrokerOperabilityMetrics(String localCluster, String brokerName) { this.metricsList = new ArrayList<>(); this.localCluster = localCluster; - this.topicLoadStats = new TopicLoadStats(); + this.topicLoadStats = new DimensionStats("topic_load_times", 60); + this.zkWriteLatencyStats = new DimensionStats("zk_write_latency", 60); + this.zkReadLatencyStats = new DimensionStats("zk_read_latency", 60); this.brokerName = brokerName; } @@ -48,25 +53,37 @@ public List getMetrics() { private void generate() { metricsList.add(getTopicLoadMetrics()); + metricsList.add(getZkWriteLatencyMetrics()); + metricsList.add(getZkReadLatencyMetrics()); } Metrics getTopicLoadMetrics() { + return getDimensionMetrics("topic_load_times", "topic_load", topicLoadStats); + } + + Metrics getZkWriteLatencyMetrics() { + return getDimensionMetrics("zk_write_latency", "zk_write_latency", zkWriteLatencyStats); + } + + Metrics getZkReadLatencyMetrics() { + return getDimensionMetrics("zk_read_latency", "zk_read_latency", zkReadLatencyStats); + } + + Metrics getDimensionMetrics(String metricsName, String dimensionName, DimensionStats stats) { Map dimensionMap = Maps.newHashMap(); dimensionMap.put("broker", brokerName); dimensionMap.put("cluster", localCluster); - dimensionMap.put("metric", "topic_load_times"); + dimensionMap.put("metric", metricsName); Metrics dMetrics = Metrics.create(dimensionMap); - topicLoadStats.updateStats(); - - dMetrics.put("brk_topic_load_time_mean_ms", topicLoadStats.meanTopicLoadMs); - dMetrics.put("brk_topic_load_time_median_ms", topicLoadStats.medianTopicLoadMs); - dMetrics.put("brk_topic_load_time_95percentile_ms", topicLoadStats.topicLoad95Ms); - dMetrics.put("brk_topic_load_time_99_percentile_ms", topicLoadStats.topicLoad99Ms); - dMetrics.put("brk_topic_load_time_99_9_percentile_ms", topicLoadStats.topicLoad999Ms); - dMetrics.put("brk_topic_load_time_99_99_percentile_ms", topicLoadStats.topicsLoad9999Ms); - dMetrics.put("brk_topic_load_rate_s", - (1000 * topicLoadStats.topicLoadCounts) / topicLoadStats.elapsedIntervalMs); + dMetrics.put("brk_" + dimensionName + "_time_mean_ms", stats.getMeanDimension()); + dMetrics.put("brk_" + dimensionName + "_time_median_ms", stats.getMedianDimension()); + dMetrics.put("brk_" + dimensionName + "_time_75percentile_ms", stats.getDimension75()); + dMetrics.put("brk_" + dimensionName + "_time_95percentile_ms", stats.getDimension95()); + dMetrics.put("brk_" + dimensionName + "_time_99_percentile_ms", stats.getDimension99()); + dMetrics.put("brk_" + dimensionName + "_time_99_9_percentile_ms", stats.getDimension999()); + dMetrics.put("brk_" + dimensionName + "_time_99_99_percentile_ms", stats.getDimension9999()); + dMetrics.put("brk_" + dimensionName + "_rate_s", stats.getDimensionCount()); return dMetrics; } @@ -76,6 +93,14 @@ public void reset() { } public void recordTopicLoadTimeValue(long topicLoadLatencyMs) { - topicLoadStats.recordTopicLoadTimeValue(topicLoadLatencyMs); + topicLoadStats.recordDimensionTimeValue(topicLoadLatencyMs, TimeUnit.MILLISECONDS); + } + + public void recordZkWriteLatencyTimeValue(long topicLoadLatencyMs) { + zkWriteLatencyStats.recordDimensionTimeValue(topicLoadLatencyMs, TimeUnit.MILLISECONDS); + } + + public void recordZkReadLatencyTimeValue(long topicLoadLatencyMs) { + zkReadLatencyStats.recordDimensionTimeValue(topicLoadLatencyMs, TimeUnit.MILLISECONDS); } } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/DimensionStats.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/DimensionStats.java new file mode 100644 index 0000000000000..08db011f6c39a --- /dev/null +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/DimensionStats.java @@ -0,0 +1,111 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.broker.stats; + +import static io.prometheus.client.CollectorRegistry.defaultRegistry; + +import java.util.concurrent.TimeUnit; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import io.prometheus.client.Collector; +import io.prometheus.client.Summary; +import io.prometheus.client.Summary.Builder; + +/** + */ +public class DimensionStats { + + private final String name; + private final String dimensionSumLabel; + private final String dimensionCountLabel; + private final Summary summary; + private static final double[] QUANTILES = { 0.50, 0.75, 0.95, 0.99, 0.999, 0.9999 }; + private static final String[] QUANTILE_LABEL = { "quantile" }; + + public DimensionStats(String name, long updateDurationInSec) { + this.name = name; + this.dimensionSumLabel = name + "_sum"; + this.dimensionCountLabel = name + "_count"; + Builder summaryBuilder = Summary.build().name(name).help("-"); + for (int i = 0; i < QUANTILES.length; i++) { + summaryBuilder.quantile(QUANTILES[i], 0.01); + } + this.summary = summaryBuilder.maxAgeSeconds(updateDurationInSec).create(); + try { + defaultRegistry.register(summary); + } catch (IllegalArgumentException ie) { + // it only happens in test-cases when try to register summary multiple times in registry + log.warn("{} is already registred {}", name, ie.getMessage()); + } + } + + public void recordDimensionTimeValue(long latency, TimeUnit unit) { + summary.observe(unit.toMillis(latency)); + } + + public double getMeanDimension() { + double sum = getDimensionSum(); + double count = getDimensionCount(); + if (!Double.isNaN(sum) && !Double.isNaN(count)) { + return sum / count; + } + return 0; + } + + public double getMedianDimension() { + return getQuantile(QUANTILES[0]); + } + + public double getDimension75() { + return getQuantile(QUANTILES[1]); + } + + public double getDimension95() { + return getQuantile(QUANTILES[2]); + } + + public double getDimension99() { + return getQuantile(QUANTILES[3]); + } + + public double getDimension999() { + return getQuantile(QUANTILES[4]); + } + + public double getDimension9999() { + return getQuantile(QUANTILES[5]); + } + + public double getDimensionSum() { + return defaultRegistry.getSampleValue(dimensionSumLabel).doubleValue(); + } + + public double getDimensionCount() { + return defaultRegistry.getSampleValue(dimensionCountLabel).doubleValue(); + } + + private double getQuantile(double q) { + return defaultRegistry.getSampleValue(name, QUANTILE_LABEL, new String[] { Collector.doubleToGoString(q) }) + .doubleValue(); + } + + private static final Logger log = LoggerFactory.getLogger(DimensionStats.class); +} diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/TopicLoadStats.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/TopicLoadStats.java deleted file mode 100644 index 7ad2e6fec6981..0000000000000 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/TopicLoadStats.java +++ /dev/null @@ -1,69 +0,0 @@ -/** - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ -package org.apache.pulsar.broker.stats; - -import org.HdrHistogram.Histogram; -import org.HdrHistogram.Recorder; - -import java.util.concurrent.TimeUnit; - -/** - */ -public class TopicLoadStats { - - /** Statistics for topic load times **/ - public double meanTopicLoadMs; - - public double medianTopicLoadMs; - - public double topicLoad95Ms; - - public double topicLoad99Ms; - - public double topicLoad999Ms; - - public double topicsLoad9999Ms; - - public double topicLoadCounts; - - public double elapsedIntervalMs; - - private Recorder topicLoadTimeRecorder = new Recorder(TimeUnit.MINUTES.toMillis(10), 2); - private Histogram topicLoadHistogram = null; - private double topicLoadRecordStartTime = TimeUnit.NANOSECONDS.toMillis(System.nanoTime()); - - public void updateStats() { - - topicLoadHistogram = topicLoadTimeRecorder.getIntervalHistogram(topicLoadHistogram); - this.elapsedIntervalMs = (TimeUnit.NANOSECONDS.toMillis(System.nanoTime()) - topicLoadRecordStartTime); - topicLoadRecordStartTime = TimeUnit.NANOSECONDS.toMillis(System.nanoTime()); - - this.meanTopicLoadMs = topicLoadHistogram.getMean(); - this.medianTopicLoadMs = topicLoadHistogram.getValueAtPercentile(50); - this.topicLoad95Ms = topicLoadHistogram.getValueAtPercentile(95); - this.topicLoad99Ms = topicLoadHistogram.getValueAtPercentile(99); - this.topicLoad999Ms = topicLoadHistogram.getValueAtPercentile(99.9); - this.topicsLoad9999Ms = topicLoadHistogram.getValueAtPercentile(99.99); - this.topicLoadCounts = topicLoadHistogram.getTotalCount(); - } - - public void recordTopicLoadTimeValue(long topicLoadLatencyMs) { - topicLoadTimeRecorder.recordValue(topicLoadLatencyMs); - } -} diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/zookeeper/aspectj/ClientCnxnAspect.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/zookeeper/aspectj/ClientCnxnAspect.java new file mode 100644 index 0000000000000..006660576ed26 --- /dev/null +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/zookeeper/aspectj/ClientCnxnAspect.java @@ -0,0 +1,181 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.broker.zookeeper.aspectj; + +import java.lang.reflect.Field; +import java.util.List; + +import org.apache.bookkeeper.util.MathUtils; +import org.apache.jute.Record; +import org.apache.zookeeper.proto.ConnectRequest; +import org.apache.zookeeper.proto.CreateRequest; +import org.apache.zookeeper.proto.DeleteRequest; +import org.apache.zookeeper.proto.ExistsRequest; +import org.apache.zookeeper.proto.GetACLRequest; +import org.apache.zookeeper.proto.GetChildren2Request; +import org.apache.zookeeper.proto.GetChildrenRequest; +import org.apache.zookeeper.proto.GetDataRequest; +import org.apache.zookeeper.proto.GetMaxChildrenRequest; +import org.apache.zookeeper.proto.GetSASLRequest; +import org.apache.zookeeper.proto.SetACLRequest; +import org.apache.zookeeper.proto.SetDataRequest; +import org.apache.zookeeper.proto.SetMaxChildrenRequest; +import org.apache.zookeeper.proto.SetSASLRequest; +import org.apache.zookeeper.proto.SetWatches; +import org.apache.zookeeper.proto.SyncRequest; +import org.aspectj.lang.ProceedingJoinPoint; +import org.aspectj.lang.annotation.Around; +import org.aspectj.lang.annotation.Aspect; +import org.aspectj.lang.annotation.Pointcut; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import jersey.repackaged.com.google.common.collect.Lists; + +@Aspect +public class ClientCnxnAspect { + + public static enum EventType { + write, read, other; + } + + public static interface EventListner { + public void recordLatency(EventType eventType, long latencyMiliSecond); + } + + private static final List listeners = Lists.newArrayList(); + + @Pointcut("execution(* org.apache.zookeeper.ClientCnxn.*.processEvent(..))") + public void processEvent() { + } + + @Around("processEvent()") + public void timedProcessEvent(ProceedingJoinPoint joinPoint) throws Throwable { + joinPoint.proceed(); + + long startTimeMs = getStartTime(joinPoint.getArgs()[0]); + if (startTimeMs == -1) { + // couldn't find start time + return; + } + Record request = getEventType(joinPoint.getArgs()[0]); + + if (request != null) { + long timeElapsed = (MathUtils.now() - startTimeMs); + notifyListeners(checkType(request), timeElapsed); + } + } + + private void notifyListeners(EventType eventType, long timeElapsed) { + listeners.forEach(listener -> { + try { + listener.recordLatency(eventType, timeElapsed); + } catch (Exception e) { + LOG.warn("Listener failed to record latency ", e); + } + }); + } + + private EventType checkType(Record response) { + + if (response == null) { + return EventType.other; + } else if (response instanceof ConnectRequest) { + return EventType.write; + } else if (response instanceof CreateRequest) { + return EventType.write; + } else if (response instanceof DeleteRequest) { + return EventType.write; + } else if (response instanceof SetDataRequest) { + return EventType.write; + } else if (response instanceof SetACLRequest) { + return EventType.write; + } else if (response instanceof SetMaxChildrenRequest) { + return EventType.write; + } else if (response instanceof SetSASLRequest) { + return EventType.write; + } else if (response instanceof SetWatches) { + return EventType.write; + } else if (response instanceof SyncRequest) { + return EventType.write; + } else if (response instanceof ExistsRequest) { + return EventType.read; + } else if (response instanceof GetDataRequest) { + return EventType.read; + } else if (response instanceof GetMaxChildrenRequest) { + return EventType.read; + } else if (response instanceof GetACLRequest) { + return EventType.read; + } else if (response instanceof GetChildrenRequest) { + return EventType.read; + } else if (response instanceof GetChildren2Request) { + return EventType.read; + } else if (response instanceof GetSASLRequest) { + return EventType.read; + } else { + return EventType.other; + } + } + + private long getStartTime(Object packet) { + try { + if (packet.getClass().getName().equals("org.apache.zookeeper.ClientCnxn$Packet")) { + Field ctxField = Class.forName("org.apache.zookeeper.ClientCnxn$Packet").getDeclaredField("ctx"); + ctxField.setAccessible(true); + Object zooworker = ctxField.get(packet); + if (zooworker.getClass().getName().equals("org.apache.bookkeeper.zookeeper.ZooWorker")) { + Field timeField = Class.forName("org.apache.bookkeeper.zookeeper.ZooWorker") + .getDeclaredField("startTimeMs"); + timeField.setAccessible(true); + long startTime = (long) timeField.get(zooworker); + return startTime; + } + } + } catch (Exception e) { + if (LOG.isDebugEnabled()) { + LOG.debug("Failed to get event-start-time from zk-response", e); + } + } + return -1; + } + + private Record getEventType(Object packet) { + try { + if (packet.getClass().getName().equals("org.apache.zookeeper.ClientCnxn$Packet")) { + Field field = Class.forName("org.apache.zookeeper.ClientCnxn$Packet").getDeclaredField("request"); + field.setAccessible(true); + Record response = (Record) field.get(packet); + return response; + } + } catch (Exception e) { + if (LOG.isDebugEnabled()) { + LOG.debug("Failed to get event-type from zk-response", e); + } + } + + return null; + } + + public static void addListener(EventListner listener) { + listeners.add(listener); + } + + private static final Logger LOG = LoggerFactory.getLogger(ClientCnxnAspect.class); + +} diff --git a/pulsar-broker/src/main/resources/META-INF/aop.xml b/pulsar-broker/src/main/resources/META-INF/aop.xml new file mode 100644 index 0000000000000..9fed26b7c54ca --- /dev/null +++ b/pulsar-broker/src/main/resources/META-INF/aop.xml @@ -0,0 +1,31 @@ + + + + + + + + + + + + \ No newline at end of file diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java index b4ba4d059dcfc..f20d4995828f7 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java @@ -320,7 +320,7 @@ public void testBrokerStatsMetrics() throws Exception { consumer.close(); Thread.sleep(ASYNC_EVENT_COMPLETION_WAIT); JsonArray metrics = brokerStatsClient.getMetrics(); - assertEquals(metrics.size(), 4, metrics.toString()); + assertEquals(metrics.size(), 6, metrics.toString()); // these metrics seem to be arriving in different order at different times... // is the order really relevant here? diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/zookeeper/ZooKeeperClientAspectJTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/zookeeper/ZooKeeperClientAspectJTest.java new file mode 100644 index 0000000000000..8c490c627ca36 --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/zookeeper/ZooKeeperClientAspectJTest.java @@ -0,0 +1,337 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.broker.zookeeper; + +import static org.mockito.Mockito.doReturn; +import static org.testng.Assert.assertNotEquals; +import static org.testng.Assert.assertTrue; + +import java.io.Closeable; +import java.io.File; +import java.io.IOException; +import java.net.InetSocketAddress; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.bookkeeper.test.PortManager; +import org.apache.pulsar.broker.PulsarService; +import org.apache.pulsar.broker.service.BrokerService; +import org.apache.pulsar.broker.service.BrokerTestBase; +import org.apache.pulsar.broker.zookeeper.aspectj.ClientCnxnAspect; +import org.apache.pulsar.broker.zookeeper.aspectj.ClientCnxnAspect.EventType; +import org.apache.pulsar.client.api.PulsarClient; +import org.apache.pulsar.common.stats.Metrics; +import org.apache.pulsar.zookeeper.LocalBookkeeperEnsemble; +import org.apache.pulsar.zookeeper.ZooKeeperClientFactory; +import org.apache.pulsar.zookeeper.ZooKeeperClientFactory.SessionType; +import org.apache.pulsar.zookeeper.ZookeeperBkClientFactoryImpl; +import org.apache.zookeeper.CreateMode; +import org.apache.zookeeper.ZooDefs; +import org.apache.zookeeper.ZooKeeper; +import org.apache.zookeeper.ZooKeeper.States; +import org.apache.zookeeper.data.ACL; +import org.apache.zookeeper.data.Stat; +import org.apache.zookeeper.server.NIOServerCnxnFactory; +import org.apache.zookeeper.server.ZooKeeperServer; +import org.aspectj.weaver.loadtime.Agent; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.testng.Assert; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.Test; + +import com.ea.agentloader.AgentLoader; +import com.google.common.util.concurrent.AtomicDouble; + +public class ZooKeeperClientAspectJTest { + + private ZookeeperServerTest localZkS; + private ZooKeeper localZkc; + private final int LOCAL_ZOOKEEPER_PORT = PortManager.nextFreePort(); + private final long ZOOKEEPER_SESSION_TIMEOUT_MILLIS = 2000; + private final List Acl = ZooDefs.Ids.OPEN_ACL_UNSAFE; + + static { + // load agent with aspectjweaver-Agent for testing + AgentLoader.loadAgentClass(Agent.class.getName(), null); + } + + @Test + public void testZkConnected() throws Exception { + try { + ZooKeeperClientFactory zkf = new ZookeeperBkClientFactoryImpl(); + CompletableFuture zkFuture = zkf.create("127.0.0.1:" + LOCAL_ZOOKEEPER_PORT, SessionType.ReadWrite, + (int) ZOOKEEPER_SESSION_TIMEOUT_MILLIS); + localZkc = zkFuture.get(ZOOKEEPER_SESSION_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS); + assertTrue(localZkc.getState().isConnected()); + assertNotEquals(localZkc.getState(), States.CONNECTEDREADONLY); + }finally{ + if (localZkc != null) { + localZkc.close(); + } + } + } + + @BeforeMethod + void setup() throws Exception { + localZkS = new ZookeeperServerTest(LOCAL_ZOOKEEPER_PORT); + localZkS.start(); + } + + @AfterMethod + void teardown() throws Exception { + localZkS.close(); + } + + /** + * Verifies that aspect-advice calculates the latency of of zk-operation + * + * @throws Exception + */ + @Test(timeOut = 7000) + void testZkClientAspectJTrigger() throws Exception { + ZooKeeperClientFactory zkf = new ZookeeperBkClientFactoryImpl(); + CompletableFuture zkFuture = zkf.create("127.0.0.1:" + LOCAL_ZOOKEEPER_PORT, SessionType.ReadWrite, + (int) ZOOKEEPER_SESSION_TIMEOUT_MILLIS); + localZkc = zkFuture.get(ZOOKEEPER_SESSION_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS); + try { + assertTrue(localZkc.getState().isConnected()); + assertNotEquals(localZkc.getState(), States.CONNECTEDREADONLY); + + final AtomicInteger writeCount = new AtomicInteger(0); + final AtomicInteger readCount = new AtomicInteger(0); + ClientCnxnAspect.addListener((EventType eventType, long latencyMiliSecond) -> { + if (eventType.equals(EventType.write)) { + writeCount.incrementAndGet(); + } else if (eventType.equals(EventType.read)) { + readCount.incrementAndGet(); + } + }); + CountDownLatch createLatch = new CountDownLatch(1); + CountDownLatch deleteLatch = new CountDownLatch(1); + CountDownLatch readLatch = new CountDownLatch(1); + CountDownLatch existLatch = new CountDownLatch(1); + localZkc.create("/createTest", "data".getBytes(), Acl, CreateMode.EPHEMERAL, (rc, path, ctx, name) -> { + createLatch.countDown(); + }, "create"); + localZkc.delete("/deleteTest", -1, (rc, path, ctx) -> { + deleteLatch.countDown(); + }, "delete"); + localZkc.exists("/createTest", null, (int rc, String path, Object ctx, Stat stat) -> { + existLatch.countDown(); + }, null); + localZkc.getData("/createTest", null, (int rc, String path, Object ctx, byte data[], Stat stat) -> { + readLatch.countDown(); + }, null); + createLatch.await(); + deleteLatch.await(); + existLatch.await(); + readLatch.await(); + Thread.sleep(500); + Assert.assertEquals(writeCount.get(), 2); + Assert.assertEquals(readCount.get(), 2); + } finally { + if (localZkc != null) { + localZkc.close(); + } + } + } + + /** + * Verifies that aspect-advice calculates the latency of of zk-operation and updates PulsarStats + * + * @throws Exception + */ + @Test(timeOut = 7000) + public void testZkOpStatsMetrics() throws Exception { + + ZooKeeperClientFactory zkf = new ZookeeperBkClientFactoryImpl(); + CompletableFuture zkFuture = zkf.create("127.0.0.1:" + LOCAL_ZOOKEEPER_PORT, SessionType.ReadWrite, + (int) ZOOKEEPER_SESSION_TIMEOUT_MILLIS); + localZkc = zkFuture.get(ZOOKEEPER_SESSION_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS); + + MockPulsar mockPulsar = new MockPulsar(localZkc); + mockPulsar.setup(); + try { + PulsarClient pulsarClient = mockPulsar.getClient(); + PulsarService pulsar = mockPulsar.getPulsar(); + + pulsarClient.createProducer("persistent://my-property/use/my-ns/my-topic1"); + Metrics zkOpMetric = getMetric(pulsar, "zk_write_latency"); + Assert.assertNotNull(zkOpMetric); + Assert.assertTrue(zkOpMetric.getMetrics().containsKey("brk_zk_write_latency_rate_s")); + Assert.assertTrue(zkOpMetric.getMetrics().containsKey("brk_zk_write_latency_time_95percentile_ms")); + Assert.assertTrue(zkOpMetric.getMetrics().containsKey("brk_zk_write_latency_time_99_99_percentile_ms")); + Assert.assertTrue(zkOpMetric.getMetrics().containsKey("brk_zk_write_latency_time_99_9_percentile_ms")); + Assert.assertTrue(zkOpMetric.getMetrics().containsKey("brk_zk_write_latency_time_99_percentile_ms")); + Assert.assertTrue(zkOpMetric.getMetrics().containsKey("brk_zk_write_latency_time_mean_ms")); + Assert.assertTrue(zkOpMetric.getMetrics().containsKey("brk_zk_write_latency_time_median_ms")); + + zkOpMetric = getMetric(pulsar, "zk_read_latency"); + Assert.assertNotNull(zkOpMetric); + Assert.assertTrue(zkOpMetric.getMetrics().containsKey("brk_zk_read_latency_rate_s")); + Assert.assertTrue(zkOpMetric.getMetrics().containsKey("brk_zk_read_latency_time_95percentile_ms")); + Assert.assertTrue(zkOpMetric.getMetrics().containsKey("brk_zk_read_latency_time_99_99_percentile_ms")); + Assert.assertTrue(zkOpMetric.getMetrics().containsKey("brk_zk_read_latency_time_99_9_percentile_ms")); + Assert.assertTrue(zkOpMetric.getMetrics().containsKey("brk_zk_read_latency_time_99_percentile_ms")); + Assert.assertTrue(zkOpMetric.getMetrics().containsKey("brk_zk_read_latency_time_mean_ms")); + Assert.assertTrue(zkOpMetric.getMetrics().containsKey("brk_zk_read_latency_time_median_ms")); + + CountDownLatch createLatch = new CountDownLatch(1); + CountDownLatch deleteLatch = new CountDownLatch(1); + CountDownLatch readLatch = new CountDownLatch(1); + CountDownLatch existLatch = new CountDownLatch(1); + localZkc.create("/createTest", "data".getBytes(), Acl, CreateMode.EPHEMERAL, (rc, path, ctx, name) -> { + createLatch.countDown(); + }, "create"); + localZkc.delete("/deleteTest", -1, (rc, path, ctx) -> { + deleteLatch.countDown(); + }, "delete"); + localZkc.exists("/createTest", null, (int rc, String path, Object ctx, Stat stat) -> { + existLatch.countDown(); + }, null); + localZkc.getData("/createTest", null, (int rc, String path, Object ctx, byte data[], Stat stat) -> { + readLatch.countDown(); + }, null); + createLatch.await(); + deleteLatch.await(); + existLatch.await(); + readLatch.await(); + Thread.sleep(500); + + BrokerService brokerService = pulsar.getBrokerService(); + brokerService.updateRates(); + List metrics = brokerService.getDestinationMetrics(); + AtomicDouble writeRate = new AtomicDouble(); + AtomicDouble readRate = new AtomicDouble(); + metrics.forEach(m -> { + if ("zk_write_latency".equalsIgnoreCase(m.getDimension("metric"))) { + writeRate.set((double) m.getMetrics().get("brk_zk_write_latency_rate_s")); + } else if ("zk_read_latency".equalsIgnoreCase(m.getDimension("metric"))) { + readRate.set((double) m.getMetrics().get("brk_zk_read_latency_rate_s")); + } + }); + Assert.assertTrue(readRate.get() > 0); + Assert.assertTrue(writeRate.get() > 0); + } finally { + mockPulsar.cleanup(); + if (localZkc != null) { + localZkc.close(); + } + } + } + + private Metrics getMetric(PulsarService pulsar, String dimension) { + BrokerService brokerService = pulsar.getBrokerService(); + brokerService.updateRates(); + for (Metrics metric : brokerService.getDestinationMetrics()) { + if (dimension.equalsIgnoreCase(metric.getDimension("metric"))) { + return metric; + } + } + return null; + } + + class ZookeeperServerTest implements Closeable { + private final File zkTmpDir; + private ZooKeeperServer zks; + private NIOServerCnxnFactory serverFactory; + private final int zkPort; + private final String hostPort; + + public ZookeeperServerTest(int zkPort) throws IOException { + this.zkPort = zkPort; + this.hostPort = "127.0.0.1:" + zkPort; + this.zkTmpDir = File.createTempFile("zookeeper", "test"); + log.info("**** Start GZK on {} ****", zkTmpDir); + if (!zkTmpDir.delete() || !zkTmpDir.mkdir()) { + throw new IOException("Couldn't create zk directory " + zkTmpDir); + } + } + + public void start() throws IOException { + try { + zks = new ZooKeeperServer(zkTmpDir, zkTmpDir, ZooKeeperServer.DEFAULT_TICK_TIME); + zks.setMaxSessionTimeout(20000); + serverFactory = new NIOServerCnxnFactory(); + serverFactory.configure(new InetSocketAddress(zkPort), 1000); + serverFactory.startup(zks); + } catch (Exception e) { + log.error("Exception while instantiating ZooKeeper", e); + } + + LocalBookkeeperEnsemble.waitForServerUp(hostPort, 30000); + log.info("ZooKeeper started at {}", hostPort); + } + + public void stop() throws IOException { + zks.shutdown(); + serverFactory.shutdown(); + log.info("Stoppend ZK server at {}", hostPort); + } + + @Override + public void close() throws IOException { + zks.shutdown(); + serverFactory.shutdown(); + zkTmpDir.delete(); + } + + private final Logger log = LoggerFactory.getLogger(ZookeeperServerTest.class); + } + + class MockPulsar extends BrokerTestBase { + + private final ZooKeeper zk; + + public MockPulsar(ZooKeeper zk) { + this.zk = zk; + } + + @Override + protected void setup() throws Exception { + super.baseSetup(); + doReturn(new ZooKeeperClientFactory() { + @Override + public CompletableFuture create(String serverList, SessionType sessionType, + int zkSessionTimeoutMillis) { + return CompletableFuture.completedFuture(zk); + } + }).when(pulsar).getZooKeeperClientFactory(); + } + + @Override + protected void cleanup() throws Exception { + super.internalCleanup(); + } + + public PulsarService getPulsar() { + return pulsar; + } + + public PulsarClient getClient() { + return pulsarClient; + } + + } +} diff --git a/pulsar-zookeeper-utils/src/main/java/org/apache/pulsar/zookeeper/ZookeeperBkClientFactoryImpl.java b/pulsar-zookeeper-utils/src/main/java/org/apache/pulsar/zookeeper/ZookeeperBkClientFactoryImpl.java new file mode 100644 index 0000000000000..f854df87cd510 --- /dev/null +++ b/pulsar-zookeeper-utils/src/main/java/org/apache/pulsar/zookeeper/ZookeeperBkClientFactoryImpl.java @@ -0,0 +1,122 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.zookeeper; + +import static com.google.common.base.Preconditions.checkArgument; + +import java.lang.reflect.Constructor; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; + +import org.apache.bookkeeper.zookeeper.BoundExponentialBackoffRetryPolicy; +import org.apache.bookkeeper.zookeeper.RetryPolicy; +import org.apache.bookkeeper.zookeeper.ZooKeeperClient; +import org.apache.bookkeeper.zookeeper.ZooKeeperWatcherBase; +import org.apache.zookeeper.KeeperException; +import org.apache.zookeeper.Watcher.Event; +import org.apache.zookeeper.ZooKeeper; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class ZookeeperBkClientFactoryImpl implements ZooKeeperClientFactory { + + @Override + public CompletableFuture create(String serverList, SessionType sessionType, int zkSessionTimeoutMillis) { + boolean canBeReadOnly = sessionType == SessionType.AllowReadOnly; + + CompletableFuture future = new CompletableFuture<>(); + try { + CompletableFuture internalFuture = new CompletableFuture<>(); + + try { + Constructor cn = (Constructor) ZooKeeperClient.class + .getDeclaredConstructor(String.class, int.class, ZooKeeperWatcherBase.class, RetryPolicy.class); + cn.setAccessible(true); + ZooKeeperWatcherBase zkWatch = new ZooKeeperWatcherBase(zkSessionTimeoutMillis); + zkWatch.addChildWatcher((event) -> { + + if (event.getType() == Event.EventType.None) { + switch (event.getState()) { + + case ConnectedReadOnly: + checkArgument(canBeReadOnly); + // Fall through + case SyncConnected: + // ZK session is ready to use + internalFuture.complete(null); + break; + + case Expired: + internalFuture + .completeExceptionally(KeeperException.create(KeeperException.Code.SESSIONEXPIRED)); + break; + + default: + log.warn("Unexpected ZK event received: {}", event); + break; + } + } + }); + final ZooKeeperClient zk = cn.newInstance(serverList, zkSessionTimeoutMillis, zkWatch, + new BoundExponentialBackoffRetryPolicy(zkSessionTimeoutMillis, zkSessionTimeoutMillis, 0)); + + internalFuture.thenRun(() -> { + log.info("ZooKeeper session established: {}", zk); + // Sometimes: ZooKeeperWatcherBase triggers childwatch before setting zk.getState() = CONNECTED and + // state is still in CONNECTING state which takes few more msec to update state CONNECTED + completeFutureOnConnect(future, zk, System.currentTimeMillis() + zkSessionTimeoutMillis, Executors.newSingleThreadScheduledExecutor()); + }).exceptionally((exception) -> { + log.error("Failed to establish ZooKeeper session: {}", exception.getMessage()); + future.completeExceptionally(exception); + return null; + }); + + internalFuture.complete(null); + } catch (Exception ex) { + future.completeExceptionally(ex); + } + + } catch (Exception e) { + future.completeExceptionally(e); + } + + return future; + } + + private void completeFutureOnConnect(CompletableFuture future, ZooKeeperClient zk, long timeOutTime, + ScheduledExecutorService executor) { + if (zk.getState().isConnected()) { + future.complete(zk); + executor.shutdown(); + } else { + if (System.currentTimeMillis() - timeOutTime < 0) { + executor.schedule(() -> completeFutureOnConnect(future, zk, timeOutTime, executor), 100, + TimeUnit.MILLISECONDS); + } else { + future.completeExceptionally( + new IllegalStateException("zookeeper couldn't connect with in given zkSessionTimeout")); + executor.shutdown(); + } + } + } + + private static final Logger log = LoggerFactory.getLogger(ZookeeperBkClientFactoryImpl.class); +} diff --git a/pulsar-zookeeper-utils/src/test/java/org/apache/pulsar/zookeeper/ZookeeperBkClientFactoryImplTest.java b/pulsar-zookeeper-utils/src/test/java/org/apache/pulsar/zookeeper/ZookeeperBkClientFactoryImplTest.java new file mode 100644 index 0000000000000..18f95d9c08759 --- /dev/null +++ b/pulsar-zookeeper-utils/src/test/java/org/apache/pulsar/zookeeper/ZookeeperBkClientFactoryImplTest.java @@ -0,0 +1,84 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.zookeeper; + +import static org.testng.Assert.assertNotEquals; +import static org.testng.Assert.assertTrue; + +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; + +import org.apache.bookkeeper.test.PortManager; +import org.apache.pulsar.zookeeper.ZookeeperBkClientFactoryImpl; +import org.apache.zookeeper.ZooKeeper; +import org.apache.zookeeper.ZooKeeper.States; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.Test; + +import org.apache.pulsar.zookeeper.ZooKeeperClientFactory.SessionType; + +@Test +public class ZookeeperBkClientFactoryImplTest { + + private ZookeeperServerTest localZkS; + private ZooKeeper localZkc; + private final int LOCAL_ZOOKEEPER_PORT = PortManager.nextFreePort(); + private final long ZOOKEEPER_SESSION_TIMEOUT_MILLIS = 1000; + + @BeforeMethod + void setup() throws Exception { + localZkS = new ZookeeperServerTest(LOCAL_ZOOKEEPER_PORT); + localZkS.start(); + } + + @AfterMethod + void teardown() throws Exception { + localZkS.close(); + } + + @Test + void testZKCreationRW() throws Exception { + ZooKeeperClientFactory zkf = new ZookeeperBkClientFactoryImpl(); + CompletableFuture zkFuture = zkf.create("127.0.0.1:" + LOCAL_ZOOKEEPER_PORT, SessionType.ReadWrite, + (int) ZOOKEEPER_SESSION_TIMEOUT_MILLIS); + localZkc = zkFuture.get(ZOOKEEPER_SESSION_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS); + assertTrue(localZkc.getState().isConnected()); + assertNotEquals(localZkc.getState(), States.CONNECTEDREADONLY); + localZkc.close(); + } + + @Test + void testZKCreationRO() throws Exception { + ZooKeeperClientFactory zkf = new ZookeeperBkClientFactoryImpl(); + CompletableFuture zkFuture = zkf.create("127.0.0.1:" + LOCAL_ZOOKEEPER_PORT, + SessionType.AllowReadOnly, (int) ZOOKEEPER_SESSION_TIMEOUT_MILLIS); + localZkc = zkFuture.get(ZOOKEEPER_SESSION_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS); + assertTrue(localZkc.getState().isConnected()); + localZkc.close(); + } + + @Test + void testZKCreationFailure() throws Exception { + ZooKeeperClientFactory zkf = new ZookeeperBkClientFactoryImpl(); + CompletableFuture zkFuture = zkf.create("invalid", SessionType.ReadWrite, + (int) ZOOKEEPER_SESSION_TIMEOUT_MILLIS); + assertTrue(zkFuture.isCompletedExceptionally()); + } +}