HADOOP-18709. Add curator based ZooKeeper communication support over …

…SSL/TLS into the common library

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

@@ -132,6 +132,20 @@ public void start() throws IOException {
    * @throws IOException If the connection cannot be started.
    */
   public void start(List<AuthInfo> authInfos) throws IOException {
+    this.start(authInfos, false);
+  }
+
+  /**
+   * Start the connection to the ZooKeeper ensemble.
+   *
+   * @param authInfos  List of authentication keys.
+   * @param sslEnabled If the connection should be SSL/TLS encrypted.
+   * @throws IOException            If the connection cannot be started.
+   */
+  public void start(List<AuthInfo> authInfos, boolean sslEnabled)
+      throws IOException{
+
+    ZKClientConfig zkClientConfig = new ZKClientConfig();
 
     // Connect to the ZooKeeper ensemble
     String zkHostPort = conf.get(CommonConfigurationKeys.ZK_ADDRESS);
@@ -171,6 +185,34 @@ public void start(List<AuthInfo> authInfos) throws IOException {
 
     this.curator = client;
   }
+  /* Check on SSL/TLS client connection requirements to emit the name of the
+   configuration missing. It improves supportability. */
+  private void validateSslConfiguration(Configuration config) throws IOException {
+    if (StringUtils.isEmpty(config.get(CommonConfigurationKeys.ZK_SSL_KEYSTORE_LOCATION))) {
+      throw new IOException(
+          "The SSL encryption is enabled for the component's ZooKeeper client connection, "
+              + "however the " + CommonConfigurationKeys.ZK_SSL_KEYSTORE_LOCATION + " " +
+              "parameter is empty.");
+    }
+    if (StringUtils.isEmpty(config.get(CommonConfigurationKeys.ZK_SSL_KEYSTORE_PASSWORD))) {
+      throw new IOException(
+          "The SSL encryption is enabled for the component's " + "ZooKeeper client connection, "
+              + "however the " + CommonConfigurationKeys.ZK_SSL_KEYSTORE_PASSWORD + " " +
+              "parameter is empty.");
+    }
+    if (StringUtils.isEmpty(config.get(CommonConfigurationKeys.ZK_SSL_TRUSTSTORE_LOCATION))) {
+      throw new IOException(
+          "The SSL encryption is enabled for the component's ZooKeeper client connection, "
+              + "however the " + CommonConfigurationKeys.ZK_SSL_TRUSTSTORE_LOCATION + " " +
+              "parameter is empty.");
+    }
+    if (StringUtils.isEmpty(config.get(CommonConfigurationKeys.ZK_SSL_TRUSTSTORE_PASSWORD))) {
+      throw new IOException(
+          "The SSL encryption is enabled for the component's ZooKeeper client connection, "
+              + "however the " + CommonConfigurationKeys.ZK_SSL_TRUSTSTORE_PASSWORD + "  " +
+              "parameter is empty.");
+    }
+  }
 
   /**
    * Get ACLs for a ZNode.
@@ -503,4 +545,44 @@ private void setJaasConfiguration(ZKClientConfig zkClientConfig) throws IOExcept
       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;
+    }
+  }
 }