Isolate zk client config for suppressing authentication

For when more than one zk servers are accessed within a single jvm,
global authentication settings are shared via system properties. This PR
bumps zk client version for getting a support for client config
isolation and makes use of it. Curator has also been updated for
compatibility.

pr-link: Alluxio#9438
change-id: cid-d9607acb78abb6550721a741de095ad2c2e2acd8

core/common/src/main/java/alluxio/PropertyKey.java

@@ -488,6 +488,13 @@ public String toString() {
           .setDescription("Session timeout to use when connecting to Zookeeper")
           .setConsistencyCheckLevel(ConsistencyCheckLevel.WARN)
           .build();
+  public static final PropertyKey ZOOKEEPER_AUTH_ENABLED =
+      new Builder(Name.ZOOKEEPER_AUTH_ENABLED)
+          .setDefaultValue(true)
+          .setDescription("If true, enable client-side Zookeeper authentication.")
+          .setConsistencyCheckLevel(ConsistencyCheckLevel.WARN)
+          .setScope(Scope.CLIENT)
+          .build();
 
   /**
    * UFS related properties.
@@ -3255,6 +3262,7 @@ public static final class Name {
         "alluxio.zookeeper.leader.inquiry.retry";
     public static final String ZOOKEEPER_LEADER_PATH = "alluxio.zookeeper.leader.path";
     public static final String ZOOKEEPER_SESSION_TIMEOUT = "alluxio.zookeeper.session.timeout";
+    public static final String ZOOKEEPER_AUTH_ENABLED = "alluxio.zookeeper.auth.enabled";
 
     //
     // UFS related properties

core/common/src/main/java/alluxio/master/MasterInquireClient.java

@@ -80,7 +80,8 @@ public static MasterInquireClient create(AlluxioConfiguration conf) {
       if (conf.getBoolean(PropertyKey.ZOOKEEPER_ENABLED)) {
         return ZkMasterInquireClient.getClient(conf.get(PropertyKey.ZOOKEEPER_ADDRESS),
             conf.get(PropertyKey.ZOOKEEPER_ELECTION_PATH),
-            conf.get(PropertyKey.ZOOKEEPER_LEADER_PATH));
+            conf.get(PropertyKey.ZOOKEEPER_LEADER_PATH),
+            conf.getBoolean(PropertyKey.ZOOKEEPER_AUTH_ENABLED));
       } else {
         return new SingleMasterInquireClient(
             NetworkAddressUtils.getConnectAddress(ServiceType.MASTER_RPC, conf));

core/common/src/main/java/alluxio/master/ZkMasterInquireClient.java

@@ -23,7 +23,10 @@
 import org.apache.curator.framework.CuratorFramework;
 import org.apache.curator.framework.CuratorFrameworkFactory;
 import org.apache.curator.retry.ExponentialBackoffRetry;
+import org.apache.curator.utils.ZookeeperFactory;
+import org.apache.zookeeper.Watcher;
 import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.client.ZKClientConfig;
 import org.apache.zookeeper.data.Stat;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -53,20 +56,42 @@ public final class ZkMasterInquireClient implements MasterInquireClient, Closeab
   private final CuratorFramework mClient;
   private final int mMaxTry;
 
+  /**
+   * Zookeeper factory for curator that controls enabling/disabling client authentication.
+   */
+  private class AlluxioZookeeperFactory implements ZookeeperFactory {
+    private boolean mAuthEnabled;
+
+    public AlluxioZookeeperFactory(boolean authEnabled) {
+      mAuthEnabled = authEnabled;
+    }
+
+    @Override
+    public ZooKeeper newZooKeeper(String connectString, int sessionTimeout, Watcher watcher,
+                                  boolean canBeReadOnly) throws Exception {
+      ZKClientConfig zkConfig = new ZKClientConfig();
+      zkConfig.setProperty(ZKClientConfig.ENABLE_CLIENT_SASL_KEY,
+          Boolean.toString(mAuthEnabled).toLowerCase());
+      return new ZooKeeper(connectString, sessionTimeout, watcher, zkConfig);
+    }
+  }
+
   /**
    * Gets the client.
    *
    * @param zookeeperAddress the address for Zookeeper
    * @param electionPath the path of the master election
    * @param leaderPath the path of the leader
+   * @param authEnabled if Alluxio client-side auth is enabled
    * @return the client
    */
   public static synchronized ZkMasterInquireClient getClient(String zookeeperAddress,
-      String electionPath, String leaderPath) {
+      String electionPath, String leaderPath, boolean authEnabled) {
     ZkMasterConnectDetails connectDetails =
         new ZkMasterConnectDetails(zookeeperAddress, leaderPath);
     if (!sCreatedClients.containsKey(connectDetails)) {
-      sCreatedClients.put(connectDetails, new ZkMasterInquireClient(connectDetails, electionPath));
+      sCreatedClients.put(connectDetails,
+          new ZkMasterInquireClient(connectDetails, electionPath, authEnabled));
     }
     return sCreatedClients.get(connectDetails);
   }
@@ -76,15 +101,19 @@ public static synchronized ZkMasterInquireClient getClient(String zookeeperAddre
    *
    * @param connectDetails connect details
    * @param electionPath the path of the master election
+   * @param authEnabled if Alluxio client-side auth is enabled
    */
-  private ZkMasterInquireClient(ZkMasterConnectDetails connectDetails, String electionPath) {
+  private ZkMasterInquireClient(ZkMasterConnectDetails connectDetails, String electionPath,
+      boolean authEnabled) {
     mConnectDetails = connectDetails;
     mElectionPath = electionPath;
 
     LOG.info("Creating new zookeeper client for {}", connectDetails);
-    // Start the client lazily.
-    mClient = CuratorFrameworkFactory.newClient(connectDetails.getZkAddress(),
-        new ExponentialBackoffRetry(Constants.SECOND_MS, 3));
+    CuratorFrameworkFactory.Builder curatorBuilder = CuratorFrameworkFactory.builder();
+    curatorBuilder.connectString(connectDetails.getZkAddress());
+    curatorBuilder.retryPolicy(new ExponentialBackoffRetry(Constants.SECOND_MS, 3));
+    curatorBuilder.zookeeperFactory(new AlluxioZookeeperFactory(authEnabled));
+    mClient = curatorBuilder.build();
 
     mMaxTry = Configuration.getInt(PropertyKey.ZOOKEEPER_LEADER_INQUIRY_RETRY_COUNT);
   }

minicluster/src/main/java/alluxio/master/MultiMasterLocalAlluxioCluster.java

@@ -21,10 +21,10 @@
 import alluxio.underfs.options.DeleteOptions;
 import alluxio.util.CommonUtils;
 import alluxio.util.WaitForOptions;
-import alluxio.zookeeper.RestartableTestingServer;
 
 import com.google.common.base.Function;
 import com.google.common.base.Throwables;
+import org.apache.curator.test.TestingServer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -41,7 +41,7 @@
 public final class MultiMasterLocalAlluxioCluster extends AbstractLocalAlluxioCluster {
   private static final Logger LOG = LoggerFactory.getLogger(MultiMasterLocalAlluxioCluster.class);
 
-  private RestartableTestingServer mCuratorServer = null;
+  private TestingServer mCuratorServer = null;
   private int mNumOfMasters = 0;
 
   private final List<LocalAlluxioMaster> mMasters = new ArrayList<>();
@@ -65,7 +65,7 @@ public MultiMasterLocalAlluxioCluster(int numMasters, int numWorkers) {
 
     try {
       mCuratorServer =
-          new RestartableTestingServer(-1, AlluxioTestDirectory.createTemporaryDirectory("zk"));
+              new TestingServer(-1, AlluxioTestDirectory.createTemporaryDirectory("zk"));
       LOG.info("Started testing zookeeper: {}", mCuratorServer.getConnectString());
     } catch (Exception e) {
       throw Throwables.propagate(e);

minicluster/src/main/java/alluxio/multi/process/MultiProcessCluster.java

@@ -37,13 +37,13 @@
 import alluxio.util.io.PathUtils;
 import alluxio.util.network.NetworkAddressUtils;
 import alluxio.wire.MasterInfo;
-import alluxio.zookeeper.RestartableTestingServer;
 
 import com.google.common.base.Function;
 import com.google.common.base.Preconditions;
 import com.google.common.io.Closer;
 import org.apache.commons.io.Charsets;
 import org.apache.commons.io.FileUtils;
+import org.apache.curator.test.TestingServer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -107,7 +107,7 @@ public final class MultiProcessCluster {
   /** Addresses of all masters. Should have the same size as {@link #mMasters}. */
   private List<MasterNetAddress> mMasterAddresses;
   private State mState;
-  private RestartableTestingServer mCuratorServer;
+  private TestingServer mCuratorServer;
   /**
    * Tracks whether the test has succeeded. If mSuccess is never updated before {@link #destroy()},
    * the state of the cluster will be saved as a tarball in the artifacts directory.
@@ -161,7 +161,7 @@ public synchronized void start() throws Exception {
         break;
       case ZOOKEEPER_HA:
         mCuratorServer = mCloser.register(
-            new RestartableTestingServer(-1, AlluxioTestDirectory.createTemporaryDirectory("zk")));
+            new TestingServer(-1, AlluxioTestDirectory.createTemporaryDirectory("zk")));
         mProperties.put(PropertyKey.ZOOKEEPER_ENABLED, "true");
         mProperties.put(PropertyKey.ZOOKEEPER_ADDRESS, mCuratorServer.getConnectString());
         break;
@@ -510,7 +510,8 @@ public synchronized MasterInquireClient getMasterInquireClient() {
       case ZOOKEEPER_HA:
         return ZkMasterInquireClient.getClient(mCuratorServer.getConnectString(),
             Configuration.get(PropertyKey.ZOOKEEPER_ELECTION_PATH),
-            Configuration.get(PropertyKey.ZOOKEEPER_LEADER_PATH));
+            Configuration.get(PropertyKey.ZOOKEEPER_LEADER_PATH),
+            Configuration.getBoolean(PropertyKey.ZOOKEEPER_AUTH_ENABLED));
       default:
         throw new IllegalStateException("Unknown deploy mode: " + mDeployMode.toString());
     }

pom.xml

@@ -130,7 +130,7 @@
   </repositories>
 
   <properties>
-    <apache.curator.version>2.1.0-incubating</apache.curator.version>
+    <apache.curator.version>4.2.0</apache.curator.version>
     <aws.amazonaws.version>1.11.215</aws.amazonaws.version>
     <build.path>build</build.path>
     <checkstyle.path>build/checkstyle/</checkstyle.path>
@@ -533,7 +533,7 @@
       <dependency>
         <groupId>org.apache.zookeeper</groupId>
         <artifactId>zookeeper</artifactId>
-        <version>3.4.6</version>
+        <version>3.5.5</version>
       </dependency>
       <dependency>
         <groupId>org.eclipse.jetty</groupId>

shell/pom.xml

@@ -57,6 +57,10 @@
       <groupId>org.reflections</groupId>
       <artifactId>reflections</artifactId>
     </dependency>
+   <dependency>
+      <groupId>jline</groupId>
+      <artifactId>jline</artifactId>
+   </dependency>
 
     <!-- Internal dependencies -->
     <dependency>