Use zookeeper-api module in Apache Helix and remove IOItec import

pinot-broker/src/test/java/org/apache/pinot/broker/util/FakePropertyStore.java

@@ -20,7 +20,7 @@
 
 import java.util.HashMap;
 import java.util.Map;
-import org.I0Itec.zkclient.IZkDataListener;
+import org.apache.helix.zookeeper.api.zkclient.IZkDataListener;
 import org.apache.helix.ZNRecord;
 import org.apache.helix.manager.zk.ZkBaseDataAccessor;
 import org.apache.helix.store.zk.ZkHelixPropertyStore;

pinot-common/src/main/java/org/apache/pinot/common/assignment/InstancePartitionsUtils.java

@@ -22,12 +22,12 @@
 import java.util.Collections;
 import java.util.List;
 import javax.annotation.Nullable;
-import org.I0Itec.zkclient.exception.ZkException;
 import org.apache.helix.AccessOption;
 import org.apache.helix.HelixManager;
 import org.apache.helix.ZNRecord;
 import org.apache.helix.store.HelixPropertyStore;
 import org.apache.helix.store.zk.ZkHelixPropertyStore;
+import org.apache.helix.zookeeper.api.zkclient.exception.ZkException;
 import org.apache.pinot.common.config.TableConfig;
 import org.apache.pinot.common.config.TableNameBuilder;
 import org.apache.pinot.common.config.TagNameUtils;

pinot-common/src/main/java/org/apache/pinot/common/metadata/ZKMetadataProvider.java

@@ -23,10 +23,10 @@
 import java.util.List;
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
-import org.I0Itec.zkclient.exception.ZkBadVersionException;
 import org.apache.helix.AccessOption;
 import org.apache.helix.ZNRecord;
 import org.apache.helix.store.zk.ZkHelixPropertyStore;
+import org.apache.helix.zookeeper.api.zkclient.exception.ZkBadVersionException;
 import org.apache.pinot.common.config.TableConfig;
 import org.apache.pinot.common.config.TableNameBuilder;
 import org.apache.pinot.spi.data.Schema;

pinot-common/src/main/java/org/apache/pinot/common/utils/ZkStarter.java

@@ -22,7 +22,10 @@
 import java.io.IOException;
 import java.net.InetSocketAddress;
 import java.util.concurrent.TimeUnit;
-import org.I0Itec.zkclient.ZkClient;
+
+import org.apache.helix.manager.zk.client.DedicatedZkClientFactory;
+import org.apache.helix.manager.zk.client.HelixZkClient;
+import org.apache.helix.zookeeper.api.zkclient.ZkClient;
 import org.apache.zookeeper.server.ServerConfig;
 import org.apache.zookeeper.server.ZooKeeperServerMain;
 import org.apache.zookeeper.server.quorum.QuorumPeerConfig;
@@ -163,7 +166,10 @@ public void run() {
       }.start();
 
       // Wait until the ZK server is started
-      ZkClient client = new ZkClient("localhost:" + port, 10000);
+      HelixZkClient.ZkConnectionConfig zkConnectionConfig =
+          new HelixZkClient.ZkConnectionConfig("localhost:" + port).setSessionTimeout(10000);
+      HelixZkClient client =
+          DedicatedZkClientFactory.getInstance().buildZkClient(zkConnectionConfig);
       client.waitUntilConnected(10L, TimeUnit.SECONDS);
       client.close();
 

pinot-common/src/main/java/org/apache/pinot/common/utils/helix/HelixHelper.java

@@ -26,7 +26,7 @@
 import java.util.Set;
 import java.util.concurrent.Callable;
 import javax.annotation.Nullable;
-import org.I0Itec.zkclient.exception.ZkBadVersionException;
+import org.apache.helix.zookeeper.api.zkclient.exception.ZkBadVersionException;
 import org.apache.helix.AccessOption;
 import org.apache.helix.HelixAdmin;
 import org.apache.helix.HelixDataAccessor;

pinot-common/src/main/java/org/apache/pinot/common/utils/helix/TableCache.java

@@ -23,8 +23,8 @@
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
-import org.I0Itec.zkclient.IZkChildListener;
-import org.I0Itec.zkclient.IZkDataListener;
+import org.apache.helix.zookeeper.api.zkclient.IZkChildListener;
+import org.apache.helix.zookeeper.api.zkclient.IZkDataListener;
 import org.apache.helix.AccessOption;
 import org.apache.helix.ZNRecord;
 import org.apache.helix.store.HelixPropertyListener;

pinot-controller/pom.xml

@@ -160,12 +160,18 @@
       </exclusions>
     </dependency>
     <dependency>
-      <groupId>io.netty</groupId>
-      <artifactId>netty</artifactId>
+      <groupId>io.dropwizard.metrics</groupId>
+      <artifactId>metrics-core</artifactId>
+      <version>3.2.3</version>
+    </dependency>
+    <dependency>
+      <groupId>org.codehaus.jackson</groupId>
+      <artifactId>jackson-mapper-asl</artifactId>
+      <version>1.9.13</version>
     </dependency>
     <dependency>
-      <groupId>com.101tec</groupId>
-      <artifactId>zkclient</artifactId>
+      <groupId>io.netty</groupId>
+      <artifactId>netty</artifactId>
     </dependency>
     <dependency>
       <groupId>org.mockito</groupId>

pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotRealtimeSegmentManager.java

@@ -27,8 +27,8 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
-import org.I0Itec.zkclient.IZkChildListener;
-import org.I0Itec.zkclient.IZkDataListener;
+import org.apache.helix.zookeeper.api.zkclient.IZkChildListener;
+import org.apache.helix.zookeeper.api.zkclient.IZkDataListener;
 import org.apache.helix.PropertyPathConfig;
 import org.apache.helix.PropertyType;
 import org.apache.helix.ZNRecord;

pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java

@@ -23,7 +23,7 @@
 import java.util.Map;
 import java.util.TreeMap;
 import java.util.concurrent.TimeoutException;
-import org.I0Itec.zkclient.exception.ZkBadVersionException;
+import org.apache.helix.zookeeper.api.zkclient.exception.ZkBadVersionException;
 import org.apache.commons.configuration.Configuration;
 import org.apache.helix.AccessOption;
 import org.apache.helix.HelixDataAccessor;

pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManagerTest.java

@@ -27,7 +27,9 @@
 import java.util.Map;
 import java.util.Random;
 import java.util.Set;
-import org.I0Itec.zkclient.ZkClient;
+
+import org.apache.helix.manager.zk.client.DedicatedZkClientFactory;
+import org.apache.helix.manager.zk.client.HelixZkClient;
 import org.apache.helix.HelixDataAccessor;
 import org.apache.helix.PropertyKey;
 import org.apache.helix.PropertyPathBuilder;
@@ -121,16 +123,20 @@ public void testGetInstanceConfigs()
       Assert.assertEquals(cachedInstanceConfig, realInstanceConfig);
     }
 
-    ZkClient zkClient = new ZkClient(_helixResourceManager.getHelixZkURL(), CONNECTION_TIMEOUT_IN_MILLISECOND,
-        CONNECTION_TIMEOUT_IN_MILLISECOND, new ZNRecordSerializer());
+    HelixZkClient.ZkConnectionConfig zkConnectionConfig =
+        new HelixZkClient.ZkConnectionConfig(_helixResourceManager.getHelixZkURL())
+            .setSessionTimeout(CONNECTION_TIMEOUT_IN_MILLISECOND);
+    HelixZkClient zkClient = DedicatedZkClientFactory.getInstance()
+        .buildZkClient(zkConnectionConfig,
+            new HelixZkClient.ZkClientConfig().setZkSerializer(new ZNRecordSerializer()));
 
     modifyExistingInstanceConfig(zkClient);
     addAndRemoveNewInstanceConfig(zkClient);
 
     zkClient.close();
   }
 
-  private void modifyExistingInstanceConfig(ZkClient zkClient)
+  private void modifyExistingInstanceConfig(HelixZkClient zkClient)
       throws InterruptedException {
     String instanceName = "Server_localhost_" + new Random().nextInt(NUM_INSTANCES);
     String instanceConfigPath = PropertyPathBuilder.instanceConfig(getHelixClusterName(), instanceName);
@@ -162,7 +168,7 @@ private void modifyExistingInstanceConfig(ZkClient zkClient)
     zkClient.writeData(instanceConfigPath, znRecord);
   }
 
-  private void addAndRemoveNewInstanceConfig(ZkClient zkClient) {
+  private void addAndRemoveNewInstanceConfig(HelixZkClient zkClient) {
     int biggerRandomNumber = NUM_INSTANCES + new Random().nextInt(NUM_INSTANCES);
     String instanceName = "Server_localhost_" + biggerRandomNumber;
     String instanceConfigPath = PropertyPathBuilder.instanceConfig(getHelixClusterName(), instanceName);

pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkQueryEngine.java

@@ -24,7 +24,9 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
-import org.I0Itec.zkclient.ZkClient;
+
+import org.apache.helix.manager.zk.client.DedicatedZkClientFactory;
+import org.apache.helix.manager.zk.client.HelixZkClient;
 import org.apache.helix.ZNRecord;
 import org.apache.helix.manager.zk.ZNRecordSerializer;
 import org.apache.pinot.broker.requesthandler.OptimizationFlags;
@@ -109,7 +111,11 @@ public void startPinot()
       _perfBenchmarkDriver.addSegment(TABLE_NAME, segmentMetadata);
     }
 
-    ZkClient client = new ZkClient("localhost:2191", 10000, 10000, new ZNRecordSerializer());
+    HelixZkClient.ZkConnectionConfig zkConnectionConfig =
+        new HelixZkClient.ZkConnectionConfig("localhost:2191")
+            .setSessionTimeout(10000);
+    HelixZkClient client = DedicatedZkClientFactory.getInstance().buildZkClient(zkConnectionConfig,
+        new HelixZkClient.ZkClientConfig().setZkSerializer(new ZNRecordSerializer()));
 
     ZNRecord record = client.readData("/PinotPerfTestCluster/EXTERNALVIEW/" + TABLE_NAME);
     while (true) {

pinot-tools/src/main/java/org/apache/pinot/tools/admin/command/StartZookeeperCommand.java

@@ -21,7 +21,7 @@
 import com.google.common.io.Files;
 import java.io.File;
 import java.io.IOException;
-import org.I0Itec.zkclient.IDefaultNameSpace;
+import org.apache.helix.zookeeper.api.zkclient.IDefaultNameSpace;
 import org.apache.commons.io.FileUtils;
 import org.apache.pinot.common.utils.ZkStarter;
 import org.apache.pinot.tools.Command;
@@ -107,7 +107,7 @@ public boolean execute()
 
     IDefaultNameSpace _defaultNameSpace = new IDefaultNameSpace() {
       @Override
-      public void createDefaultNameSpace(org.I0Itec.zkclient.ZkClient zkClient) {
+      public void createDefaultNameSpace(org.apache.helix.zookeeper.api.zkclient.ZkClient zkClient) {
         // init any zk paths if needed
       }
     };

pinot-tools/src/main/java/org/apache/pinot/tools/perf/ZookeeperLauncher.java

@@ -20,8 +20,8 @@
 
 import com.google.common.base.Preconditions;
 import java.io.File;
-import org.I0Itec.zkclient.IDefaultNameSpace;
-import org.I0Itec.zkclient.ZkServer;
+import org.apache.helix.zookeeper.api.zkclient.IDefaultNameSpace;
+import org.apache.helix.zookeeper.api.zkclient.ZkServer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -53,7 +53,7 @@ public ZookeeperLauncher(String baseTempDir) {
   public boolean start(int zkPort) {
     IDefaultNameSpace defaultNameSpace = new IDefaultNameSpace() {
       @Override
-      public void createDefaultNameSpace(org.I0Itec.zkclient.ZkClient zkClient) {
+      public void createDefaultNameSpace(org.apache.helix.zookeeper.api.zkclient.ZkClient zkClient) {
         // init any zk paths if needed
       }
     };

pom.xml

@@ -114,7 +114,7 @@
     <!-- Configuration for unit/integration tests section 1 of 3 (properties) ENDS HERE.-->
     <avro.version>1.7.6</avro.version>
     <parquet.version>1.8.0</parquet.version>
-    <helix.version>0.9.4</helix.version>
+    <helix.version>0.9.2-SNAPSHOT</helix.version>
     <zkclient.version>0.7</zkclient.version>
     <jackson.version>2.9.8</jackson.version>
     <async-http-client.version>1.9.21</async-http-client.version>
@@ -495,6 +495,29 @@
           </exclusion>
         </exclusions>
       </dependency>
+      <dependency>
+        <groupId>org.apache.helix</groupId>
+        <artifactId>zookeeper-api</artifactId>
+        <version>${helix.version}</version>
+        <exclusions>
+          <exclusion>
+            <groupId>com.github.sgroschupf</groupId>
+            <artifactId>zkclient</artifactId>
+          </exclusion>
+          <exclusion>
+            <groupId>org.jboss.netty</groupId>
+            <artifactId>netty</artifactId>
+          </exclusion>
+          <exclusion>
+            <groupId>log4j</groupId>
+            <artifactId>log4j</artifactId>
+          </exclusion>
+          <exclusion>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-log4j12</artifactId>
+          </exclusion>
+        </exclusions>
+      </dependency>
 
       <!-- log4j2 related dependencies -->
       <dependency>