Removed AspectJ based metrics for ZooKeeper

bin/pulsar

@@ -197,17 +197,11 @@ add_maven_deps_to_classpath() {
 }
 
 if [ -d "$PULSAR_HOME/lib" ]; then
-	PULSAR_CLASSPATH=$PULSAR_CLASSPATH:$PULSAR_HOME/lib/*
-    ASPECTJ_AGENT_PATH=`ls -1 $PULSAR_HOME/lib/org.aspectj-aspectjweaver-*.jar`
+	  PULSAR_CLASSPATH=$PULSAR_CLASSPATH:$PULSAR_HOME/lib/*
 else
     add_maven_deps_to_classpath
-
-    ASPECTJ_VERSION=`grep '<aspectj.version>' $PULSAR_HOME/pom.xml | awk -F'>' '{print $2}' | awk -F'<' '{print $1}'`
-    ASPECTJ_AGENT_PATH="$HOME/.m2/repository/org/aspectj/aspectjweaver/$ASPECTJ_VERSION/aspectjweaver-$ASPECTJ_VERSION.jar"
 fi
 
-ASPECTJ_AGENT="-javaagent:$ASPECTJ_AGENT_PATH"
-
 # if no args specified, show usage
 if [ $# = 0 ]; then
     pulsar_help;
@@ -321,23 +315,23 @@ LOG4J2_SHUTDOWN_HOOK_DISABLED="-Dlog4j.shutdownHookEnabled=false"
 cd "$PULSAR_HOME"
 if [ $COMMAND == "broker" ]; then
     PULSAR_LOG_FILE=${PULSAR_LOG_FILE:-"pulsar-broker.log"}
-    exec $JAVA $LOG4J2_SHUTDOWN_HOOK_DISABLED $OPTS $ASPECTJ_AGENT -Dpulsar.log.file=$PULSAR_LOG_FILE org.apache.pulsar.PulsarBrokerStarter --broker-conf $PULSAR_BROKER_CONF $@
+    exec $JAVA $LOG4J2_SHUTDOWN_HOOK_DISABLED $OPTS -Dpulsar.log.file=$PULSAR_LOG_FILE org.apache.pulsar.PulsarBrokerStarter --broker-conf $PULSAR_BROKER_CONF $@
 elif [ $COMMAND == "bookie" ]; then
     PULSAR_LOG_FILE=${PULSAR_LOG_FILE:-"bookkeeper.log"}
     exec $JAVA $OPTS -Dpulsar.log.file=$PULSAR_LOG_FILE org.apache.bookkeeper.server.Main --conf $PULSAR_BOOKKEEPER_CONF $@
 elif [ $COMMAND == "zookeeper" ]; then
     PULSAR_LOG_FILE=${PULSAR_LOG_FILE:-"zookeeper.log"}
-    exec $JAVA ${ZK_OPTS} $OPTS $ASPECTJ_AGENT -Dpulsar.log.file=$PULSAR_LOG_FILE org.apache.pulsar.zookeeper.ZooKeeperStarter $PULSAR_ZK_CONF $@
+    exec $JAVA ${ZK_OPTS} $OPTS -Dpulsar.log.file=$PULSAR_LOG_FILE org.apache.pulsar.zookeeper.ZooKeeperStarter $PULSAR_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
     OPTS="${OPTS} ${ZK_OPTS} -Dreadonlymode.enabled=true"
-    exec $JAVA $OPTS $ASPECTJ_AGENT -Dpulsar.log.file=$PULSAR_LOG_FILE org.apache.pulsar.zookeeper.ConfigurationStoreStarter $PULSAR_GLOBAL_ZK_CONF $@
+    exec $JAVA $OPTS -Dpulsar.log.file=$PULSAR_LOG_FILE org.apache.pulsar.zookeeper.ConfigurationStoreStarter $PULSAR_GLOBAL_ZK_CONF $@
 elif [ $COMMAND == "configuration-store" ]; then
     PULSAR_LOG_FILE=${PULSAR_LOG_FILE:-"configuration-store.log"}
     # Allow global ZK to turn into read-only mode when it cannot reach the quorum
     OPTS="${OPTS} ${ZK_OPTS} -Dreadonlymode.enabled=true"
-    exec $JAVA $OPTS $ASPECTJ_AGENT -Dpulsar.log.file=$PULSAR_LOG_FILE org.apache.pulsar.zookeeper.ConfigurationStoreStarter $PULSAR_CONFIGURATION_STORE_CONF $@
+    exec $JAVA $OPTS -Dpulsar.log.file=$PULSAR_LOG_FILE org.apache.pulsar.zookeeper.ConfigurationStoreStarter $PULSAR_CONFIGURATION_STORE_CONF $@
 elif [ $COMMAND == "discovery" ]; then
     PULSAR_LOG_FILE=${PULSAR_LOG_FILE:-"discovery.log"}
     exec $JAVA $OPTS -Dpulsar.log.file=$PULSAR_LOG_FILE org.apache.pulsar.discovery.service.server.DiscoveryServiceStarter $PULSAR_DISCOVERY_CONF $@
@@ -352,7 +346,7 @@ elif [ $COMMAND == "functions-worker" ]; then
     exec $JAVA $OPTS -Dpulsar.log.file=$PULSAR_LOG_FILE org.apache.pulsar.functions.worker.FunctionWorkerStarter -c $PULSAR_WORKER_CONF $@
 elif [ $COMMAND == "standalone" ]; then
     PULSAR_LOG_FILE=${PULSAR_LOG_FILE:-"pulsar-standalone.log"}
-    exec $JAVA $LOG4J2_SHUTDOWN_HOOK_DISABLED $OPTS $ASPECTJ_AGENT ${ZK_OPTS} -Dpulsar.log.file=$PULSAR_LOG_FILE org.apache.pulsar.PulsarStandaloneStarter --config $PULSAR_STANDALONE_CONF $@
+    exec $JAVA $LOG4J2_SHUTDOWN_HOOK_DISABLED $OPTS ${ZK_OPTS} -Dpulsar.log.file=$PULSAR_LOG_FILE org.apache.pulsar.PulsarStandaloneStarter --config $PULSAR_STANDALONE_CONF $@
 elif [ $COMMAND == "initialize-cluster-metadata" ]; then
     exec $JAVA $OPTS org.apache.pulsar.PulsarClusterMetadataSetup $@
 elif [ $COMMAND == "delete-cluster-metadata" ]; then

distribution/server/src/assemble/LICENSE.bin.txt

@@ -410,7 +410,6 @@ The Apache Software License, Version 2.0
     - org.apache.bookkeeper.http-vertx-http-server-4.13.0.jar
     - org.apache.bookkeeper.stats-bookkeeper-stats-api-4.13.0.jar
     - org.apache.bookkeeper.stats-prometheus-metrics-provider-4.13.0.jar
-    - org.apache.bookkeeper.tests-stream-storage-tests-common-4.13.0.jar
     - org.apache.distributedlog-distributedlog-common-4.13.0.jar
     - org.apache.distributedlog-distributedlog-core-4.13.0-tests.jar
     - org.apache.distributedlog-distributedlog-core-4.13.0.jar
@@ -509,6 +508,7 @@ The Apache Software License, Version 2.0
     - io.vertx-vertx-core-3.5.4.jar
     - io.vertx-vertx-web-3.5.4.jar
   * Apache ZooKeeper
+    - org.apache.zookeeper-zookeeper-3.6.2.jar
     - org.apache.zookeeper-zookeeper-jute-3.6.2.jar
   * Snappy Java
     - org.xerial.snappy-snappy-java-1.1.7.jar
@@ -574,11 +574,6 @@ Eclipse Distribution License 1.0 -- licenses/LICENSE-EDL-1.0.txt
    - jakarta.activation-jakarta.activation-api-1.2.1.jar
  * Jakarta XML Binding -- jakarta.xml.bind-jakarta.xml.bind-api-2.3.2.jar
 
-Eclipse Public License 1.0 -- licenses/LICENSE-AspectJ.txt
- * AspectJ
-    - org.aspectj-aspectjrt-1.9.2.jar
-    - org.aspectj-aspectjweaver-1.9.2.jar
-
 Eclipse Public License - v2.0 -- licenses/LICENSE-EPL-2.0.txt
  * Jakarta Annotations API -- jakarta.annotation-jakarta.annotation-api-1.3.5.jar
  * Jakarta RESTful Web Services -- jakarta.ws.rs-jakarta.ws.rs-api-2.1.6.jar

distribution/server/src/assemble/bin.xml

@@ -130,13 +130,6 @@
       <excludes>
         <exclude>org.apache.pulsar:pulsar-functions-runtime-all</exclude>
 
-        <!-- Already included in pulsar-zookeeper instrumented jar -->
-        <exclude>org.apache.zookeeper:zookeeper</exclude>
-
-        <!-- Explicitely remove JUnit which is getting pulled in even
-             though it's set to the scope 'test' -->
-        <exclude>junit:junit</exclude>
-
         <exclude>org.projectlombok:lombok</exclude>
 
         <!-- prevent adding pulsar-functions-api-examples in lib -->

pom.xml

@@ -114,7 +114,6 @@ flexible messaging model and an intuitive client API.</description>
     <jersey.version>2.31</jersey.version>
     <athenz.version>1.10.9</athenz.version>
     <prometheus.version>0.5.0</prometheus.version>
-    <aspectj.version>1.9.2</aspectj.version>
     <vertx.version>3.5.4</vertx.version>
     <rocksdb.version>6.10.2</rocksdb.version>
     <slf4j.version>1.7.25</slf4j.version>
@@ -214,7 +213,6 @@ flexible messaging model and an intuitive client API.</description>
     <!-- Plugin dependencies -->
     <protobuf-maven-plugin.version>0.6.1</protobuf-maven-plugin.version>
     <exec-maven-plugin.version>3.0.0</exec-maven-plugin.version>
-    <aspectj-maven-plugin.version>1.11.1</aspectj-maven-plugin.version>
     <license-maven-plugin.version>4.0.rc2</license-maven-plugin.version>
     <directory-maven-plugin.version>0.3.1</directory-maven-plugin.version>
     <maven-enforcer-plugin.version>3.0.0-M3</maven-enforcer-plugin.version>
@@ -849,18 +847,6 @@ flexible messaging model and an intuitive client API.</description>
         <version>${jsonwebtoken.version}</version>
       </dependency>
 
-      <dependency>
-        <groupId>org.aspectj</groupId>
-        <artifactId>aspectjrt</artifactId>
-        <version>${aspectj.version}</version>
-      </dependency>
-
-      <dependency>
-        <groupId>org.aspectj</groupId>
-        <artifactId>aspectjweaver</artifactId>
-        <version>${aspectj.version}</version>
-      </dependency>
-
       <dependency>
         <groupId>net.jodah</groupId>
         <artifactId>typetools</artifactId>

pulsar-broker-shaded/pom.xml

@@ -145,7 +145,6 @@
                   <include>org.yaml:snakeyaml</include>
                   <include>org.hdrhistogram:*</include>
                   <include>com.github.zafarkhaja:java-semver</include>
-                  <include>org.aspectj:*</include>
                   <include>org.apache.avro:avro</include>
                   <!-- Avro transitive dependencies-->
                   <include>com.thoughtworks.paranamer:paranamer</include>
@@ -307,14 +306,6 @@
                   <pattern>org.HdrHistogram</pattern>
                   <shadedPattern>org.apache.pulsar.shade.org.HdrHistogram</shadedPattern>
                 </relocation>
-                <relocation>
-                  <pattern>org.aspectj</pattern>
-                  <shadedPattern>org.apache.pulsar.shade.org.aspectj</shadedPattern>
-                </relocation>
-                <relocation>
-                  <pattern>aj</pattern>
-                  <shadedPattern>org.apache.pulsar.shade.aj</shadedPattern>
-                </relocation>
                 <relocation>
                   <pattern>com.ea</pattern>
                   <shadedPattern>org.apache.pulsar.shade.com.ea</shadedPattern>

pulsar-broker/pom.xml

@@ -640,19 +640,6 @@
               <configuration>
                 <lifecycleMappingMetadata>
                   <pluginExecutions>
-                    <pluginExecution>
-                      <pluginExecutionFilter>
-                        <groupId>org.codehaus.mojo</groupId>
-                        <artifactId>aspectj-maven-plugin</artifactId>
-                        <versionRange>[1.10,)</versionRange>
-                        <goals>
-                          <goal>compile</goal>
-                        </goals>
-                      </pluginExecutionFilter>
-                      <action>
-                        <ignore></ignore>
-                      </action>
-                    </pluginExecution>
                     <pluginExecution>
                       <pluginExecutionFilter>
                         <groupId>org.apache.maven.plugins</groupId>

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java

@@ -119,8 +119,6 @@
 import org.apache.pulsar.broker.stats.prometheus.metrics.Summary;
 import org.apache.pulsar.broker.systopic.SystemTopicClient;
 import org.apache.pulsar.broker.web.PulsarWebResource;
-import org.apache.pulsar.broker.zookeeper.aspectj.ClientCnxnAspect;
-import org.apache.pulsar.broker.zookeeper.aspectj.ClientCnxnAspect.EventListner;
 import org.apache.pulsar.client.admin.PulsarAdmin;
 import org.apache.pulsar.client.admin.PulsarAdminBuilder;
 import org.apache.pulsar.client.api.ClientBuilder;
@@ -244,7 +242,6 @@ public class BrokerService implements Closeable, ZooKeeperCacheListener<Policies
 
     private final int keepAliveIntervalSeconds;
     private final PulsarStats pulsarStats;
-    private final EventListner zkStatsListener;
     private final AuthenticationService authenticationService;
 
     public static final String BROKER_SERVICE_CONFIGURATION_PATH = "/admin/configuration";
@@ -353,9 +350,6 @@ public Map<String, String> deserialize(String key, byte[] content) throws Except
                     pulsar.getConfiguration().getMaxUnackedMessagesPerSubscriptionOnBrokerBlocked());
         }
 
-        // register listener to capture zk-latency
-        zkStatsListener = (eventType, latencyMs) -> pulsarStats.recordZkLatencyTimeValue(eventType, latencyMs);
-
         this.delayedDeliveryTrackerFactory = DelayedDeliveryTrackerLoader
                 .loadDelayedDeliveryTrackerFactory(pulsar.getConfiguration());
 
@@ -468,9 +462,6 @@ public void start() throws Exception {
         this.updateBrokerPublisherThrottlingMaxRate();
         this.startCheckReplicationPolicies();
         this.startDeduplicationSnapshotMonitor();
-        // register listener to capture zk-latency
-        ClientCnxnAspect.addListener(zkStatsListener);
-        ClientCnxnAspect.registerExecutor(pulsar.getExecutor());
     }
 
     protected void startStatsUpdater(int statsUpdateInitailDelayInSecs, int statsUpdateFrequencyInSecs) {
@@ -717,8 +708,6 @@ public CompletableFuture<Void> closeAsync() {
                                     log.warn("Error in closing authenticationService", e);
                                 }
                                 pulsarStats.close();
-                                ClientCnxnAspect.removeListener(zkStatsListener);
-                                ClientCnxnAspect.registerExecutor(null);
                                 try {
                                     delayedDeliveryTrackerFactory.close();
                                 } catch (IOException e) {

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarStats.java

@@ -34,7 +34,6 @@
 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;
@@ -237,18 +236,6 @@ public void recordTopicLoadTimeValue(String topic, long topicLoadLatencyMs) {
         }
     }
 
-    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());
-        }
-    }
-
     public void recordConnectionCreate() {
         brokerOperabilityMetrics.recordConnectionCreate();
     }

pulsar-broker/src/test/java/org/apache/pulsar/broker/zookeeper/ClusterMetadataSetupTest.java

@@ -18,16 +18,24 @@
  */
 package org.apache.pulsar.broker.zookeeper;
 
+import lombok.extern.slf4j.Slf4j;
 import org.apache.pulsar.PulsarClusterMetadataSetup;
-import org.apache.pulsar.broker.zookeeper.ZooKeeperClientAspectJTest.ZookeeperServerTest;
+import org.apache.pulsar.zookeeper.LocalBookkeeperEnsemble;
 import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.server.NIOServerCnxnFactory;
+import org.apache.zookeeper.server.ZooKeeperServer;
 import org.testng.annotations.AfterMethod;
 import org.testng.annotations.BeforeMethod;
 import org.testng.annotations.Test;
 
+import java.io.Closeable;
+import java.io.File;
+import java.io.IOException;
+import java.net.InetSocketAddress;
 import static org.testng.Assert.assertNotNull;
 import static org.testng.Assert.assertNull;
 
+@Slf4j
 @Test(groups = "broker")
 public class ClusterMetadataSetupTest {
     private ZookeeperServerTest localZkS;
@@ -112,4 +120,53 @@ void teardown() throws Exception {
         localZkS.close();
     }
 
+    static 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();
+        }
+
+        public int getZookeeperPort() {
+            return serverFactory.getLocalPort();
+        }
+    }
 }