HADOOP-18919. Zookeeper SSL/TLS support in HDFS ZKFC (#6194)

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ha/ActiveStandbyElector.java

@@ -29,8 +29,10 @@ import java.util.concurrent.locks.ReentrantLock;
 import org.apache.hadoop.HadoopIllegalArgumentException;
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.security.SecurityUtil;
 import org.apache.hadoop.util.ZKUtil.ZKAuthInfo;
 import org.apache.hadoop.util.StringUtils;
+import org.apache.zookeeper.client.ZKClientConfig;
 import org.apache.zookeeper.data.ACL;
 import org.apache.zookeeper.KeeperException;
 import org.apache.zookeeper.Watcher;
@@ -48,6 +50,10 @@ import org.apache.hadoop.util.Preconditions;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import javax.naming.ConfigurationException;
+
+import org.apache.hadoop.security.SecurityUtil.TruststoreKeystore;
+
 /**
  * 
  * This class implements a simple library to perform leader election on top of
@@ -170,6 +176,7 @@ public class ActiveStandbyElector implements StatCallback, StringCallback {
   private final int zkSessionTimeout;
   private final List<ACL> zkAcl;
   private final List<ZKAuthInfo> zkAuthInfo;
+  private TruststoreKeystore truststoreKeystore;
   private byte[] appData;
   private final String zkLockFilePath;
   private final String zkBreadCrumbPath;
@@ -209,6 +216,7 @@ public class ActiveStandbyElector implements StatCallback, StringCallback {
    * @param app
    *          reference to callback interface object
    * @param maxRetryNum maxRetryNum.
+   * @param truststoreKeystore truststore keystore, that we will use for ZK if SSL/TLS is enabled
    * @throws IOException raised on errors performing I/O.
    * @throws HadoopIllegalArgumentException
    *         if valid data is not supplied.
@@ -218,10 +226,10 @@ public class ActiveStandbyElector implements StatCallback, StringCallback {
   public ActiveStandbyElector(String zookeeperHostPorts,
       int zookeeperSessionTimeout, String parentZnodeName, List<ACL> acl,
       List<ZKAuthInfo> authInfo, ActiveStandbyElectorCallback app,
-      int maxRetryNum) throws IOException, HadoopIllegalArgumentException,
-      KeeperException {
+      int maxRetryNum, TruststoreKeystore truststoreKeystore)
+          throws IOException, HadoopIllegalArgumentException, KeeperException {
     this(zookeeperHostPorts, zookeeperSessionTimeout, parentZnodeName, acl,
-      authInfo, app, maxRetryNum, true);
+            authInfo, app, maxRetryNum, true, truststoreKeystore);
   }
 
   /**
@@ -254,6 +262,7 @@ public class ActiveStandbyElector implements StatCallback, StringCallback {
    * @param failFast
    *          whether need to add the retry when establishing ZK connection.
    * @param maxRetryNum max Retry Num
+   * @param truststoreKeystore truststore keystore, that we will use for ZK if SSL/TLS is enabled
    * @throws IOException
    *          raised on errors performing I/O.
    * @throws HadoopIllegalArgumentException
@@ -264,7 +273,7 @@ public class ActiveStandbyElector implements StatCallback, StringCallback {
   public ActiveStandbyElector(String zookeeperHostPorts,
       int zookeeperSessionTimeout, String parentZnodeName, List<ACL> acl,
       List<ZKAuthInfo> authInfo, ActiveStandbyElectorCallback app,
-      int maxRetryNum, boolean failFast) throws IOException,
+      int maxRetryNum, boolean failFast, TruststoreKeystore truststoreKeystore) throws IOException,
       HadoopIllegalArgumentException, KeeperException {
     if (app == null || acl == null || parentZnodeName == null
         || zookeeperHostPorts == null || zookeeperSessionTimeout <= 0) {
@@ -279,6 +288,7 @@ public class ActiveStandbyElector implements StatCallback, StringCallback {
     zkLockFilePath = znodeWorkingDir + "/" + LOCK_FILENAME;
     zkBreadCrumbPath = znodeWorkingDir + "/" + BREADCRUMB_FILENAME;
     this.maxRetryNum = maxRetryNum;
+    this.truststoreKeystore = truststoreKeystore;
 
     // establish the ZK Connection for future API calls
     if (failFast) {
@@ -740,7 +750,19 @@ public class ActiveStandbyElector implements StatCallback, StringCallback {
    * @throws IOException raised on errors performing I/O.
    */
   protected ZooKeeper createZooKeeper() throws IOException {
-    return new ZooKeeper(zkHostPort, zkSessionTimeout, watcher);
+    ZKClientConfig zkClientConfig = new ZKClientConfig();
+    if (truststoreKeystore != null) {
+      try {
+        SecurityUtil.setSslConfiguration(zkClientConfig, truststoreKeystore);
+      } catch (ConfigurationException ce) {
+        throw new IOException(ce);
+      }
+    }
+    return initiateZookeeper(zkClientConfig);
+  }
+
+  protected ZooKeeper initiateZookeeper(ZKClientConfig zkClientConfig) throws IOException {
+    return new ZooKeeper(zkHostPort, zkSessionTimeout, watcher, zkClientConfig);
   }
 
   private void fatalError(String errorMessage) {

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ha/ZKFailoverController.java

@@ -59,6 +59,8 @@ import org.apache.hadoop.thirdparty.com.google.common.util.concurrent.ThreadFact
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.apache.hadoop.security.SecurityUtil.TruststoreKeystore;
+
 @InterfaceAudience.LimitedPrivate("HDFS")
 public abstract class ZKFailoverController {
 
@@ -147,6 +149,7 @@ public abstract class ZKFailoverController {
   protected abstract InetSocketAddress getRpcAddressToBindTo();
   protected abstract PolicyProvider getPolicyProvider();
   protected abstract List<HAServiceTarget> getAllOtherNodes();
+  protected abstract boolean isSSLEnabled();
 
   /**
    * Return the name of a znode inside the configured parent znode in which
@@ -372,9 +375,10 @@ public abstract class ZKFailoverController {
     int maxRetryNum = conf.getInt(
         CommonConfigurationKeys.HA_FC_ELECTOR_ZK_OP_RETRIES_KEY,
         CommonConfigurationKeys.HA_FC_ELECTOR_ZK_OP_RETRIES_DEFAULT);
+    TruststoreKeystore truststoreKeystore = isSSLEnabled() ? new TruststoreKeystore(conf) : null;
     elector = new ActiveStandbyElector(zkQuorum,
         zkTimeout, getParentZnode(), zkAcls, zkAuths,
-        new ElectorCallbacks(), maxRetryNum);
+        new ElectorCallbacks(), maxRetryNum, truststoreKeystore);
   }
   
   private String getParentZnode() {

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/SecurityUtil.java

@@ -35,6 +35,7 @@ import java.util.ServiceLoader;
 import java.util.concurrent.TimeUnit;
 
 import javax.annotation.Nullable;
+import javax.naming.ConfigurationException;
 import javax.security.auth.kerberos.KerberosPrincipal;
 import javax.security.auth.kerberos.KerberosTicket;
 
@@ -53,6 +54,8 @@ import org.apache.hadoop.security.token.TokenInfo;
 import org.apache.hadoop.util.StopWatch;
 import org.apache.hadoop.util.StringUtils;
 import org.apache.hadoop.util.ZKUtil;
+import org.apache.zookeeper.client.ZKClientConfig;
+import org.apache.zookeeper.common.ClientX509Util;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.xbill.DNS.Name;
@@ -786,4 +789,103 @@ public final class SecurityUtil {
       throw e;
     }
   }
+
+  public static void validateSslConfiguration(TruststoreKeystore truststoreKeystore)
+          throws ConfigurationException {
+    if (org.apache.commons.lang3.StringUtils.isEmpty(truststoreKeystore.keystoreLocation)) {
+      throw new ConfigurationException(
+          "The keystore location parameter is empty for the ZooKeeper client connection.");
+    }
+    if (org.apache.commons.lang3.StringUtils.isEmpty(truststoreKeystore.keystorePassword)) {
+      throw new ConfigurationException(
+          "The keystore password parameter is empty for the ZooKeeper client connection.");
+    }
+    if (org.apache.commons.lang3.StringUtils.isEmpty(truststoreKeystore.truststoreLocation)) {
+      throw new ConfigurationException(
+          "The truststore location parameter is empty for the ZooKeeper client connection.");
+    }
+    if (org.apache.commons.lang3.StringUtils.isEmpty(truststoreKeystore.truststorePassword)) {
+      throw new ConfigurationException(
+          "The truststore password parameter is empty for the ZooKeeper client connection.");
+    }
+  }
+
+  /**
+   * Configure ZooKeeper Client with SSL/TLS connection.
+   * @param zkClientConfig ZooKeeper Client configuration
+   * @param truststoreKeystore truststore keystore, that we use to set the SSL configurations
+   * @throws ConfigurationException if the SSL configs are empty
+   */
+  public static void setSslConfiguration(ZKClientConfig zkClientConfig,
+                                         TruststoreKeystore truststoreKeystore)
+          throws ConfigurationException {
+    setSslConfiguration(zkClientConfig, truststoreKeystore, new ClientX509Util());
+  }
+
+  public static void setSslConfiguration(ZKClientConfig zkClientConfig,
+                                         TruststoreKeystore truststoreKeystore,
+                                         ClientX509Util x509Util)
+          throws ConfigurationException {
+    validateSslConfiguration(truststoreKeystore);
+    LOG.info("Configuring the ZooKeeper client to use SSL/TLS encryption for connecting to the "
+        + "ZooKeeper server.");
+    LOG.debug("Configuring the ZooKeeper client with {} location: {}.",
+        truststoreKeystore.keystoreLocation,
+        CommonConfigurationKeys.ZK_SSL_KEYSTORE_LOCATION);
+    LOG.debug("Configuring the ZooKeeper client with {} location: {}.",
+        truststoreKeystore.truststoreLocation,
+        CommonConfigurationKeys.ZK_SSL_TRUSTSTORE_LOCATION);
+
+    zkClientConfig.setProperty(ZKClientConfig.SECURE_CLIENT, "true");
+    zkClientConfig.setProperty(ZKClientConfig.ZOOKEEPER_CLIENT_CNXN_SOCKET,
+        "org.apache.zookeeper.ClientCnxnSocketNetty");
+    zkClientConfig.setProperty(x509Util.getSslKeystoreLocationProperty(),
+        truststoreKeystore.keystoreLocation);
+    zkClientConfig.setProperty(x509Util.getSslKeystorePasswdProperty(),
+        truststoreKeystore.keystorePassword);
+    zkClientConfig.setProperty(x509Util.getSslTruststoreLocationProperty(),
+        truststoreKeystore.truststoreLocation);
+    zkClientConfig.setProperty(x509Util.getSslTruststorePasswdProperty(),
+        truststoreKeystore.truststorePassword);
+  }
+
+  /**
+   * Helper class to contain the Truststore/Keystore paths for the ZK client connection over
+   * SSL/TLS.
+   */
+  public static class TruststoreKeystore {
+    private final String keystoreLocation;
+    private final String keystorePassword;
+    private final String truststoreLocation;
+    private final String truststorePassword;
+
+    /**
+     * Configuration for the ZooKeeper connection when SSL/TLS is enabled.
+     * When a value is not configured, ensure that empty string is set instead of null.
+     *
+     * @param conf ZooKeeper Client configuration
+     */
+    public TruststoreKeystore(Configuration conf) {
+      keystoreLocation = conf.get(CommonConfigurationKeys.ZK_SSL_KEYSTORE_LOCATION, "");
+      keystorePassword = conf.get(CommonConfigurationKeys.ZK_SSL_KEYSTORE_PASSWORD, "");
+      truststoreLocation = conf.get(CommonConfigurationKeys.ZK_SSL_TRUSTSTORE_LOCATION, "");
+      truststorePassword = conf.get(CommonConfigurationKeys.ZK_SSL_TRUSTSTORE_PASSWORD, "");
+    }
+
+    public String getKeystoreLocation() {
+      return keystoreLocation;
+    }
+
+    public String getKeystorePassword() {
+      return keystorePassword;
+    }
+
+    public String getTruststoreLocation() {
+      return truststoreLocation;
+    }
+
+    public String getTruststorePassword() {
+      return truststorePassword;
+    }
+  }
 }

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/curator/ZKCuratorManager.java

@@ -40,7 +40,6 @@ import org.apache.zookeeper.CreateMode;
 import org.apache.zookeeper.Watcher;
 import org.apache.zookeeper.ZooKeeper;
 import org.apache.zookeeper.client.ZKClientConfig;
-import org.apache.zookeeper.common.ClientX509Util;
 import org.apache.zookeeper.data.ACL;
 import org.apache.zookeeper.data.Stat;
 
@@ -49,7 +48,7 @@ import org.slf4j.LoggerFactory;
 
 import org.apache.hadoop.util.Preconditions;
 
-import javax.naming.ConfigurationException;
+import org.apache.hadoop.security.SecurityUtil.TruststoreKeystore;
 
 /**
  * Helper class that provides utility methods specific to ZK operations.
@@ -570,64 +569,12 @@ public final class ZKCuratorManager {
         setJaasConfiguration(zkClientConfig);
       }
       if (sslEnabled) {
-        setSslConfiguration(zkClientConfig);
+        SecurityUtil.setSslConfiguration(zkClientConfig, truststoreKeystore);
       }
       return new ZooKeeper(connectString, sessionTimeout, watcher,
           canBeReadOnly, zkClientConfig);
     }
 
-    /**
-     * Configure ZooKeeper Client with SSL/TLS connection.
-     * @param zkClientConfig ZooKeeper Client configuration
-     */
-    private void setSslConfiguration(ZKClientConfig zkClientConfig) throws ConfigurationException {
-      this.setSslConfiguration(zkClientConfig, new ClientX509Util());
-    }
-
-    private void setSslConfiguration(ZKClientConfig zkClientConfig, ClientX509Util x509Util)
-        throws ConfigurationException {
-      validateSslConfiguration();
-      LOG.info("Configuring the ZooKeeper client to use SSL/TLS encryption for connecting to the "
-          + "ZooKeeper server.");
-      LOG.debug("Configuring the ZooKeeper client with {} location: {}.",
-          this.truststoreKeystore.keystoreLocation,
-          CommonConfigurationKeys.ZK_SSL_KEYSTORE_LOCATION);
-      LOG.debug("Configuring the ZooKeeper client with {} location: {}.",
-          this.truststoreKeystore.truststoreLocation,
-          CommonConfigurationKeys.ZK_SSL_TRUSTSTORE_LOCATION);
-
-      zkClientConfig.setProperty(ZKClientConfig.SECURE_CLIENT, "true");
-      zkClientConfig.setProperty(ZKClientConfig.ZOOKEEPER_CLIENT_CNXN_SOCKET,
-          "org.apache.zookeeper.ClientCnxnSocketNetty");
-      zkClientConfig.setProperty(x509Util.getSslKeystoreLocationProperty(),
-          this.truststoreKeystore.keystoreLocation);
-      zkClientConfig.setProperty(x509Util.getSslKeystorePasswdProperty(),
-          this.truststoreKeystore.keystorePassword);
-      zkClientConfig.setProperty(x509Util.getSslTruststoreLocationProperty(),
-          this.truststoreKeystore.truststoreLocation);
-      zkClientConfig.setProperty(x509Util.getSslTruststorePasswdProperty(),
-          this.truststoreKeystore.truststorePassword);
-    }
-
-    private void validateSslConfiguration() throws ConfigurationException {
-      if (StringUtils.isEmpty(this.truststoreKeystore.keystoreLocation)) {
-        throw new ConfigurationException(
-            "The keystore location parameter is empty for the ZooKeeper client connection.");
-      }
-      if (StringUtils.isEmpty(this.truststoreKeystore.keystorePassword)) {
-        throw new ConfigurationException(
-            "The keystore password parameter is empty for the ZooKeeper client connection.");
-      }
-      if (StringUtils.isEmpty(this.truststoreKeystore.truststoreLocation)) {
-        throw new ConfigurationException(
-            "The truststore location parameter is empty for the ZooKeeper client connection.");
-      }
-      if (StringUtils.isEmpty(this.truststoreKeystore.truststorePassword)) {
-        throw new ConfigurationException(
-            "The truststore password parameter is empty for the ZooKeeper client connection.");
-      }
-    }
-
     private boolean isJaasConfigurationSet(ZKClientConfig zkClientConfig) {
       String clientConfig = zkClientConfig.getProperty(ZKClientConfig.LOGIN_CONTEXT_NAME_KEY,
           ZKClientConfig.LOGIN_CONTEXT_NAME_KEY_DEFAULT);
@@ -649,44 +596,4 @@ public final class ZKCuratorManager {
       zkClientConfig.setProperty(ZKClientConfig.LOGIN_CONTEXT_NAME_KEY, JAAS_CLIENT_ENTRY);
     }
   }
-
-  /**
-   * Helper class to contain the Truststore/Keystore paths for the ZK client connection over
-   * SSL/TLS.
-   */
-  public static class TruststoreKeystore {
-    private final String keystoreLocation;
-    private final String keystorePassword;
-    private final String truststoreLocation;
-    private final String truststorePassword;
-
-    /**
-     * Configuration for the ZooKeeper connection when SSL/TLS is enabled.
-     * When a value is not configured, ensure that empty string is set instead of null.
-     *
-     * @param conf ZooKeeper Client configuration
-     */
-    public TruststoreKeystore(Configuration conf) {
-      keystoreLocation = conf.get(CommonConfigurationKeys.ZK_SSL_KEYSTORE_LOCATION, "");
-      keystorePassword = conf.get(CommonConfigurationKeys.ZK_SSL_KEYSTORE_PASSWORD, "");
-      truststoreLocation = conf.get(CommonConfigurationKeys.ZK_SSL_TRUSTSTORE_LOCATION, "");
-      truststorePassword = conf.get(CommonConfigurationKeys.ZK_SSL_TRUSTSTORE_PASSWORD, "");
-    }
-
-    public String getKeystoreLocation() {
-      return keystoreLocation;
-    }
-
-    public String getKeystorePassword() {
-      return keystorePassword;
-    }
-
-    public String getTruststoreLocation() {
-      return truststoreLocation;
-    }
-
-    public String getTruststorePassword() {
-      return truststorePassword;
-    }
-  }
 }

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ha/MiniZKFCCluster.java

@@ -370,5 +370,10 @@ public class MiniZKFCCluster {
       }
       return services;
     }
+
+    @Override
+    protected boolean isSSLEnabled() {
+      return false;
+    }
   }
 }

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ha/TestActiveStandbyElector.java

@@ -22,6 +22,8 @@ import java.io.IOException;
 import java.util.Collections;
 import java.util.List;
 
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.security.SecurityUtil;
 import org.apache.zookeeper.CreateMode;
 import org.apache.zookeeper.KeeperException;
 import org.apache.zookeeper.KeeperException.Code;
@@ -29,12 +31,15 @@ import org.apache.zookeeper.WatchedEvent;
 import org.apache.zookeeper.Watcher;
 import org.apache.zookeeper.ZooKeeper;
 import org.apache.zookeeper.Watcher.Event;
+import org.apache.zookeeper.client.ZKClientConfig;
+import org.apache.zookeeper.common.ClientX509Util;
 import org.apache.zookeeper.data.ACL;
 import org.apache.zookeeper.data.Stat;
 import org.apache.zookeeper.ZooDefs.Ids;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.Assert;
+import org.mockito.ArgumentCaptor;
 import org.mockito.Mockito;
 
 import org.apache.hadoop.HadoopIllegalArgumentException;
@@ -63,7 +68,7 @@ public class TestActiveStandbyElector {
         KeeperException {
       super(hostPort, timeout, parent, acl, Collections
           .<ZKAuthInfo> emptyList(), app,
-          CommonConfigurationKeys.HA_FC_ELECTOR_ZK_OP_RETRIES_DEFAULT);
+          CommonConfigurationKeys.HA_FC_ELECTOR_ZK_OP_RETRIES_DEFAULT, null);
     }
 
     @Override
@@ -777,7 +782,7 @@ public class TestActiveStandbyElector {
     try {
       new ActiveStandbyElector("127.0.0.1", 2000, ZK_PARENT_NAME,
           Ids.OPEN_ACL_UNSAFE, Collections.<ZKAuthInfo> emptyList(), mockApp,
-          CommonConfigurationKeys.HA_FC_ELECTOR_ZK_OP_RETRIES_DEFAULT) {
+          CommonConfigurationKeys.HA_FC_ELECTOR_ZK_OP_RETRIES_DEFAULT, null) {
 
           @Override
           protected ZooKeeper createZooKeeper() throws IOException {
@@ -809,4 +814,85 @@ public class TestActiveStandbyElector {
     Mockito.verify(mockZK, Mockito.times(0)).create(ZK_LOCK_NAME, null,
         Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL, elector, mockZK);
   }
+
+  /**
+   * We want to test if we create an ActiveStandbyElector with null as a TruststoreKeystore,
+   * then we are creating a ZooKeeper without the SSL configs in ActiveStandbyElector and the other
+   * configs are the same as the default values.
+   * We do this by checking the ZKClientConfig properties.
+   * @throws Exception
+   */
+  @Test
+  public void testWithoutTruststoreKeystore() throws Exception {
+    ZKClientConfig defaultConfig = new ZKClientConfig();
+    ClientX509Util clientX509Util = new ClientX509Util();
+    System.out.println(defaultConfig.getProperty(ZKClientConfig.ZOOKEEPER_CLIENT_CNXN_SOCKET));
+    ActiveStandbyElector e = Mockito.spy(new ActiveStandbyElector("localhost", 1, "",
+            Collections.emptyList(), null, Mockito.mock(ActiveStandbyElectorCallback.class),
+            1, null) {
+      @Override
+      protected synchronized ZooKeeper connectToZooKeeper() {
+        return null;
+      }
+    });
+
+    e.createZooKeeper();
+
+    ArgumentCaptor<ZKClientConfig> configArgumentCaptor
+            = ArgumentCaptor.forClass(ZKClientConfig.class);
+    Mockito.verify(e).initiateZookeeper(configArgumentCaptor.capture());
+    ZKClientConfig clientConfig = configArgumentCaptor.getValue();
+    Assert.assertEquals(defaultConfig.getProperty(ZKClientConfig.SECURE_CLIENT),
+            clientConfig.getProperty(ZKClientConfig.SECURE_CLIENT));
+    Assert.assertEquals(defaultConfig.getProperty(ZKClientConfig.ZOOKEEPER_CLIENT_CNXN_SOCKET),
+            clientConfig.getProperty(ZKClientConfig.ZOOKEEPER_CLIENT_CNXN_SOCKET));
+    Assert.assertNull(clientConfig.getProperty(clientX509Util.getSslKeystoreLocationProperty()));
+    Assert.assertNull(clientConfig.getProperty(clientX509Util.getSslKeystorePasswdProperty()));
+    Assert.assertNull(clientConfig.getProperty(clientX509Util.getSslTruststoreLocationProperty()));
+    Assert.assertNull(clientConfig.getProperty(clientX509Util.getSslTruststorePasswdProperty()));
+  }
+
+  /**
+   * We want to test if we create an ActiveStandbyElector with a TruststoreKeystore, which already
+   * has the SSL configuration set, then we are creating a ZooKeeper with the correct SSL configs
+   * in ActiveStandbyElector. We do this by checking the ZKClientConfig properties.
+   * @throws Exception
+   */
+  @Test
+  public void testWithTruststoreKeystore() throws Exception {
+    Configuration conf = new Configuration();
+    ClientX509Util clientX509Util = new ClientX509Util();
+    conf.set(CommonConfigurationKeys.ZK_SSL_KEYSTORE_LOCATION, "keystore_location");
+    conf.set(CommonConfigurationKeys.ZK_SSL_KEYSTORE_PASSWORD, "keystore_password");
+    conf.set(CommonConfigurationKeys.ZK_SSL_TRUSTSTORE_LOCATION, "truststore_location");
+    conf.set(CommonConfigurationKeys.ZK_SSL_TRUSTSTORE_PASSWORD, "truststore_password");
+    SecurityUtil.TruststoreKeystore truststoreKeystore = new SecurityUtil.TruststoreKeystore(conf);
+    ActiveStandbyElector e = Mockito.spy(new ActiveStandbyElector("localhost", 1, "",
+            Collections.emptyList(), null, Mockito.mock(ActiveStandbyElectorCallback.class),
+            1, truststoreKeystore) {
+      @Override
+      protected synchronized ZooKeeper connectToZooKeeper() {
+        return null;
+      }
+    });
+
+    e.createZooKeeper();
+
+    ArgumentCaptor<ZKClientConfig> configArgumentCaptor
+            = ArgumentCaptor.forClass(ZKClientConfig.class);
+    Mockito.verify(e).initiateZookeeper(configArgumentCaptor.capture());
+    ZKClientConfig clientConfig = configArgumentCaptor.getValue();
+    Assert.assertEquals("true", clientConfig.getProperty(ZKClientConfig.SECURE_CLIENT));
+    Assert.assertEquals("org.apache.zookeeper.ClientCnxnSocketNetty",
+            clientConfig.getProperty(ZKClientConfig.ZOOKEEPER_CLIENT_CNXN_SOCKET));
+    Assert.assertEquals("keystore_location",
+            clientConfig.getProperty(clientX509Util.getSslKeystoreLocationProperty()));
+    Assert.assertEquals("keystore_password",
+            clientConfig.getProperty(clientX509Util.getSslKeystorePasswdProperty()));
+    Assert.assertEquals("truststore_location",
+            clientConfig.getProperty(clientX509Util.getSslTruststoreLocationProperty()));
+    Assert.assertEquals("truststore_password",
+            clientConfig.getProperty(clientX509Util.getSslTruststorePasswdProperty()));
+
+  }
 }

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ha/TestActiveStandbyElectorRealZK.java

@@ -70,7 +70,7 @@ public class TestActiveStandbyElectorRealZK extends ClientBaseWithFixes {
       appDatas[i] = Ints.toByteArray(i);
       electors[i] = new ActiveStandbyElector(hostPort, 5000, PARENT_DIR,
           Ids.OPEN_ACL_UNSAFE, Collections.<ZKAuthInfo> emptyList(), cbs[i],
-          CommonConfigurationKeys.HA_FC_ELECTOR_ZK_OP_RETRIES_DEFAULT);
+          CommonConfigurationKeys.HA_FC_ELECTOR_ZK_OP_RETRIES_DEFAULT, null);
     }
   }
   
@@ -270,7 +270,7 @@ public class TestActiveStandbyElectorRealZK extends ClientBaseWithFixes {
     ActiveStandbyElector elector =
         new ActiveStandbyElector(hostPort, 5000, PARENT_DIR,
             Ids.READ_ACL_UNSAFE, Collections.<ZKAuthInfo>emptyList(), cb,
-            CommonConfigurationKeys.HA_FC_ELECTOR_ZK_OP_RETRIES_DEFAULT);
+            CommonConfigurationKeys.HA_FC_ELECTOR_ZK_OP_RETRIES_DEFAULT, null);
 
     // Simulate the case by pre-creating znode 'parentZnodeName'. Then updates
     // znode's data so that data version will be increased to 1. Here znode's

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/util/curator/TestSecureZKCuratorManager.java

@@ -22,6 +22,7 @@ import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.Map;
 
+import org.apache.hadoop.security.SecurityUtil;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
@@ -146,7 +147,7 @@ public class TestSecureZKCuratorManager {
     // Validate that HadoopZooKeeperFactory will set ZKConfig with given principals
     ZKCuratorManager.HadoopZookeeperFactory factory =
         new ZKCuratorManager.HadoopZookeeperFactory(null, null, null, true,
-            new ZKCuratorManager.TruststoreKeystore(hadoopConf));
+            new SecurityUtil.TruststoreKeystore(hadoopConf));
     ZooKeeper zk = factory.newZooKeeper(this.server.getConnectString(), 1000, null, false);
     validateSSLConfiguration(this.hadoopConf.get(CommonConfigurationKeys.ZK_SSL_KEYSTORE_LOCATION),
         this.hadoopConf.get(CommonConfigurationKeys.ZK_SSL_KEYSTORE_PASSWORD),
@@ -183,8 +184,8 @@ public class TestSecureZKCuratorManager {
       Validate that the null values are converted into empty strings by the class.
      */
     Configuration conf = new Configuration();
-    ZKCuratorManager.TruststoreKeystore truststoreKeystore =
-        new ZKCuratorManager.TruststoreKeystore(conf);
+    SecurityUtil.TruststoreKeystore truststoreKeystore =
+        new SecurityUtil.TruststoreKeystore(conf);
 
     assertEquals("Validate that null value is converted to empty string.", "",
         truststoreKeystore.getKeystoreLocation());
@@ -200,8 +201,8 @@ public class TestSecureZKCuratorManager {
     conf.set(CommonConfigurationKeys.ZK_SSL_KEYSTORE_PASSWORD, "keystorePassword");
     conf.set(CommonConfigurationKeys.ZK_SSL_TRUSTSTORE_LOCATION, "/truststore.jks");
     conf.set(CommonConfigurationKeys.ZK_SSL_TRUSTSTORE_PASSWORD, "truststorePassword");
-    ZKCuratorManager.TruststoreKeystore truststoreKeystore1 =
-        new ZKCuratorManager.TruststoreKeystore(conf);
+    SecurityUtil.TruststoreKeystore truststoreKeystore1 =
+        new SecurityUtil.TruststoreKeystore(conf);
     assertEquals("Validate that non-null value kept intact.", "/keystore.jks",
         truststoreKeystore1.getKeystoreLocation());
     assertEquals("Validate that null value is converted to empty string.", "keystorePassword",

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/util/curator/TestZKCuratorManager.java

@@ -34,6 +34,7 @@ import org.apache.curator.retry.RetryNTimes;
 import org.apache.curator.test.TestingServer;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.CommonConfigurationKeys;
+import org.apache.hadoop.security.SecurityUtil;
 import org.apache.hadoop.security.authentication.util.JaasConfiguration;
 import org.apache.hadoop.util.ZKUtil;
 import org.apache.zookeeper.CreateMode;
@@ -220,7 +221,7 @@ public class TestZKCuratorManager {
         .authorization(new ArrayList<>())
         .zookeeperFactory(new ZKCuratorManager.HadoopZookeeperFactory(
             "foo1", "bar1", "bar1.keytab", false,
-            new ZKCuratorManager.TruststoreKeystore(conf))
+            new SecurityUtil.TruststoreKeystore(conf))
 
         ).build();
     client.start();

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java

@@ -1343,6 +1343,9 @@ public class DFSConfigKeys extends CommonConfigurationKeys {
   public static final int DFS_HA_ZKFC_PORT_DEFAULT = 8019;
   public static final String DFS_HA_ZKFC_NN_HTTP_TIMEOUT_KEY = "dfs.ha.zkfc.nn.http.timeout.ms";
   public static final int DFS_HA_ZKFC_NN_HTTP_TIMEOUT_KEY_DEFAULT = 20000;
+  /** Enable Zookeeper SSL/TLS communication. */
+  public static final String ZK_CLIENT_SSL_ENABLED = "dfs.ha.zkfc.client.ssl.enabled";
+  public static final boolean DEFAULT_ZK_CLIENT_SSL_ENABLED = false;
   public static final String DFS_HA_NN_NOT_BECOME_ACTIVE_IN_SAFEMODE =
       "dfs.ha.nn.not-become-active-in-safemode";
   public static final boolean DFS_HA_NN_NOT_BECOME_ACTIVE_IN_SAFEMODE_DEFAULT =

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/DFSZKFailoverController.java

@@ -294,4 +294,11 @@ public class DFSZKFailoverController extends ZKFailoverController {
     }
     return targets;
   }
+
+  @Override
+  protected boolean isSSLEnabled() {
+    return conf.getBoolean(
+        DFSConfigKeys.ZK_CLIENT_SSL_ENABLED,
+        DFSConfigKeys.DEFAULT_ZK_CLIENT_SSL_ENABLED);
+  }
 }

hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml

@@ -3756,6 +3756,14 @@
   </description>
 </property>
 
+<property>
+  <name>dfs.ha.zkfc.client.ssl.enabled</name>
+  <value>false</value>
+  <description>
+    Enable SSL/TLS encryption for the ZooKeeper communication from ZKFC.
+  </description>
+</property>
+
 <property>
   <name>dfs.ha.nn.not-become-active-in-safemode</name>
   <value>false</value>

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ActiveStandbyElectorBasedElectorService.java

@@ -18,6 +18,7 @@
 package org.apache.hadoop.yarn.server.resourcemanager;
 
 import org.apache.hadoop.classification.VisibleForTesting;
+import org.apache.hadoop.security.SecurityUtil;
 import org.apache.hadoop.thirdparty.protobuf.InvalidProtocolBufferException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -104,8 +105,12 @@ public class ActiveStandbyElectorBasedElectorService extends AbstractService
         conf.getInt(YarnConfiguration.RM_HA_FC_ELECTOR_ZK_RETRIES_KEY, conf
           .getInt(CommonConfigurationKeys.HA_FC_ELECTOR_ZK_OP_RETRIES_KEY,
             CommonConfigurationKeys.HA_FC_ELECTOR_ZK_OP_RETRIES_DEFAULT));
+    boolean isSSLEnabled = conf.getBoolean(YarnConfiguration.RM_ZK_CLIENT_SSL_ENABLED,
+            YarnConfiguration.DEFAULT_RM_ZK_CLIENT_SSL_ENABLED);
+    SecurityUtil.TruststoreKeystore truststoreKeystore
+            = isSSLEnabled ? new SecurityUtil.TruststoreKeystore(conf) : null;
     elector = new ActiveStandbyElector(zkQuorum, (int) zkSessionTimeout,
-        electionZNode, zkAcls, zkAuths, this, maxRetryNum, false);
+        electionZNode, zkAcls, zkAuths, this, maxRetryNum, false, truststoreKeystore);
 
     elector.ensureParentZNode();
     if (!isParentZnodeSafe(clusterId)) {