diff --git a/common/pom.xml b/common/pom.xml
index 973310657c44..42d7e77cc3f0 100644
--- a/common/pom.xml
+++ b/common/pom.xml
@@ -51,14 +51,26 @@
${commons-cli.version}
- org.apache.commons
- commons-lang3
- ${commons-lang3.version}
+ org.apache.commons
+ commons-lang3
+ ${commons-lang3.version}
+
+
+ org.apache.curator
+ curator-framework
+
+
+ org.apache.curator
+ curator-recipes
org.apache.orc
orc-core
+
+ org.apache.zookeeper
+ zookeeper
+
jline
jline
diff --git a/common/src/java/org/apache/hadoop/hive/common/SSLZookeeperFactory.java b/common/src/java/org/apache/hadoop/hive/common/SSLZookeeperFactory.java
new file mode 100644
index 000000000000..70602d4c6625
--- /dev/null
+++ b/common/src/java/org/apache/hadoop/hive/common/SSLZookeeperFactory.java
@@ -0,0 +1,89 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hive.common;
+
+import java.security.KeyStore;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.curator.utils.ZookeeperFactory;
+import org.apache.zookeeper.ClientCnxnSocketNetty;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.client.ZKClientConfig;
+import org.apache.zookeeper.common.ClientX509Util;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Factory to create ZooKeeper clients with TLS enabled.
+ */
+public class SSLZookeeperFactory implements ZookeeperFactory {
+ private static final Logger LOG = LoggerFactory.getLogger(SSLZookeeperFactory.class);
+
+ private final boolean sslEnabled;
+ private final String keyStoreLocation;
+ private final String keyStorePassword;
+ private final String keyStoreType;
+ private final String trustStoreLocation;
+ private final String trustStorePassword;
+ private final String trustStoreType;
+
+ public SSLZookeeperFactory(boolean sslEnabled, String keyStoreLocation, String keyStorePassword,
+ String keyStoreType, String trustStoreLocation, String trustStorePassword,
+ String trustStoreType) {
+ this.sslEnabled = sslEnabled;
+ this.keyStoreLocation = StringUtils.defaultString(keyStoreLocation, "");
+ this.keyStorePassword = StringUtils.defaultString(keyStorePassword, "");
+ this.keyStoreType = StringUtils.isBlank(keyStoreType) ? KeyStore.getDefaultType() : keyStoreType;
+ this.trustStoreLocation = StringUtils.defaultString(trustStoreLocation, "");
+ this.trustStorePassword = StringUtils.defaultString(trustStorePassword, "");
+ this.trustStoreType = StringUtils.isBlank(trustStoreType) ? KeyStore.getDefaultType() : trustStoreType;
+ if (sslEnabled) {
+ if (StringUtils.isBlank(keyStoreLocation)) {
+ LOG.warn("Missing ZooKeeper keystore location");
+ }
+ if (StringUtils.isBlank(trustStoreLocation)) {
+ LOG.warn("Missing ZooKeeper truststore location");
+ }
+ }
+ }
+
+ @Override
+ public ZooKeeper newZooKeeper(String connectString, int sessionTimeout, Watcher watcher,
+ boolean canBeReadOnly) throws Exception {
+ if (!sslEnabled) {
+ return new ZooKeeper(connectString, sessionTimeout, watcher, canBeReadOnly);
+ }
+
+ ZKClientConfig clientConfig = new ZKClientConfig();
+ clientConfig.setProperty(ZKClientConfig.SECURE_CLIENT, "true");
+ clientConfig.setProperty(ZKClientConfig.ZOOKEEPER_CLIENT_CNXN_SOCKET,
+ ClientCnxnSocketNetty.class.getName());
+
+ ClientX509Util x509Util = new ClientX509Util();
+ clientConfig.setProperty(x509Util.getSslKeystoreLocationProperty(), keyStoreLocation);
+ clientConfig.setProperty(x509Util.getSslKeystorePasswdProperty(), keyStorePassword);
+ clientConfig.setProperty(x509Util.getSslKeystoreTypeProperty(), keyStoreType);
+ clientConfig.setProperty(x509Util.getSslTruststoreLocationProperty(), trustStoreLocation);
+ clientConfig.setProperty(x509Util.getSslTruststorePasswdProperty(), trustStorePassword);
+ clientConfig.setProperty(x509Util.getSslTruststoreTypeProperty(), trustStoreType);
+
+ return new ZooKeeper(connectString, sessionTimeout, watcher, canBeReadOnly, clientConfig);
+ }
+}
diff --git a/common/src/java/org/apache/hadoop/hive/common/ZKDeRegisterWatcher.java b/common/src/java/org/apache/hadoop/hive/common/ZKDeRegisterWatcher.java
new file mode 100644
index 000000000000..22316648ab51
--- /dev/null
+++ b/common/src/java/org/apache/hadoop/hive/common/ZKDeRegisterWatcher.java
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hive.common;
+
+import org.apache.zookeeper.WatchedEvent;
+import org.apache.zookeeper.Watcher;
+
+/**
+ * The watcher class which sets the de-register flag when the given znode is deleted.
+ */
+public class ZKDeRegisterWatcher implements Watcher {
+ private final ZooKeeperHiveHelper zooKeeperHiveHelper;
+
+ public ZKDeRegisterWatcher(ZooKeeperHiveHelper zooKeeperHiveHelper) {
+ this.zooKeeperHiveHelper = zooKeeperHiveHelper;
+ }
+
+ @Override
+ public void process(WatchedEvent event) {
+ if (event.getType().equals(Watcher.Event.EventType.NodeDeleted)) {
+ zooKeeperHiveHelper.deregisterZnode();
+ }
+ }
+}
diff --git a/common/src/java/org/apache/hadoop/hive/common/ZooKeeperHiveHelper.java b/common/src/java/org/apache/hadoop/hive/common/ZooKeeperHiveHelper.java
new file mode 100644
index 000000000000..0e9d1fd456f6
--- /dev/null
+++ b/common/src/java/org/apache/hadoop/hive/common/ZooKeeperHiveHelper.java
@@ -0,0 +1,304 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hive.common;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.concurrent.TimeUnit;
+import java.util.List;
+import org.apache.curator.framework.api.ACLProvider;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.recipes.nodes.PersistentEphemeralNode;
+import org.apache.curator.retry.ExponentialBackoffRetry;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.KeeperException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+// The class serves three purposes (for HiveServer2 and HiveMetaStore)
+// 1. An instance of this class holds ZooKeeper related configuration parameter values from Hive
+// configuration and metastore configuration.
+// 2. For a server which is added to ZooKeeper specified by the configuration, an instance of
+// this class holds the znode corresponding to that server, zookeeper client used to watch the
+// znode.
+// 3. For a metastore client it provides API to find server URIs from specified ZooKeeper.
+//
+// We could have differentiated these three functionality into three different classes by
+// including an instance of first class in the second and the third, but there's isn't much stuff
+// in the first and the third.. Also note that the third functionality overlaps with
+// ZooKeeperHiveClientHelper class, but that overlap is very small. So for now all the three
+// functionality are bundled in a single class.
+
+/**
+ * ZooKeeperHiveHelper. A helper class to hold ZooKeeper related configuration, to register and
+ * deregister ZooKeeper node for a given server and to fetch registered server URIs for clients.
+ */
+public class ZooKeeperHiveHelper {
+ public static final Logger LOG = LoggerFactory.getLogger(ZooKeeperHiveHelper.class.getName());
+ public static final String ZOOKEEPER_PATH_SEPARATOR = "/";
+
+ private final String quorum;
+ private final String rootNamespace;
+ private boolean deregisteredWithZooKeeper = false; // Set to true only when deregistration happens
+ private final int connectionTimeout;
+ private final int sessionTimeout;
+ private final int baseSleepTime;
+ private final int maxRetries;
+ private final boolean sslEnabled;
+ private final String keyStoreLocation;
+ private final String keyStorePassword;
+ private final String keyStoreType;
+ private final String trustStoreLocation;
+ private final String trustStorePassword;
+ private final String trustStoreType;
+
+ private CuratorFramework zooKeeperClient;
+ private PersistentEphemeralNode znode;
+
+ public ZooKeeperHiveHelper(String quorum, String clientPort, String rootNamespace,
+ int sessionTimeout, int baseSleepTime, int maxRetries) {
+ this(quorum, clientPort, rootNamespace, 0, sessionTimeout, baseSleepTime, maxRetries);
+ }
+
+ public ZooKeeperHiveHelper(String quorum, String clientPort, String rootNamespace,
+ int connectionTimeout, int sessionTimeout, int baseSleepTime,
+ int maxRetries) {
+ this(quorum, clientPort, rootNamespace, connectionTimeout, sessionTimeout, baseSleepTime,
+ maxRetries, false, null, null, null, null, null, null);
+ }
+
+ public ZooKeeperHiveHelper(String quorum, String clientPort, String rootNamespace,
+ int connectionTimeout, int sessionTimeout, int baseSleepTime,
+ int maxRetries, boolean sslEnabled, String keyStoreLocation,
+ String keyStorePassword, String keyStoreType,
+ String trustStoreLocation, String trustStorePassword,
+ String trustStoreType) {
+ // Get the ensemble server addresses in the format host1:port1, host2:port2, ... . Append
+ // the configured port to hostname if the hostname doesn't contain a port.
+ String[] hosts = quorum.split(",");
+ StringBuilder quorumServers = new StringBuilder();
+ for (int i = 0; i < hosts.length; i++) {
+ quorumServers.append(hosts[i].trim());
+ if (!hosts[i].contains(":") && clientPort != null && !clientPort.trim().isEmpty()) {
+ quorumServers.append(":");
+ quorumServers.append(clientPort);
+ }
+
+ if (i != hosts.length - 1) {
+ quorumServers.append(",");
+ }
+ }
+
+ this.quorum = quorumServers.toString();
+ this.rootNamespace = rootNamespace;
+ this.connectionTimeout = connectionTimeout;
+ this.sessionTimeout = sessionTimeout;
+ this.baseSleepTime = baseSleepTime;
+ this.maxRetries = maxRetries;
+ this.sslEnabled = sslEnabled;
+ this.keyStoreLocation = keyStoreLocation;
+ this.keyStorePassword = keyStorePassword;
+ this.keyStoreType = keyStoreType;
+ this.trustStoreLocation = trustStoreLocation;
+ this.trustStorePassword = trustStorePassword;
+ this.trustStoreType = trustStoreType;
+ }
+
+ /**
+ * Get the ensemble server addresses. The format is: host1:port, host2:port..
+ **/
+ public String getQuorumServers() {
+ return quorum;
+ }
+
+ /**
+ * Adds a server instance to ZooKeeper as a znode.
+ *
+ * @throws Exception
+ */
+ public void addServerInstanceToZooKeeper(String znodePathPrefix, String znodeData,
+ ACLProvider zooKeeperAclProvider,
+ ZKDeRegisterWatcher watcher) throws Exception {
+ // This might be the first server getting added to the ZooKeeper, so the parent node may need
+ // to be created.
+ zooKeeperClient = startZookeeperClient(zooKeeperAclProvider, true);
+
+ // Create a znode under the rootNamespace parent for the given path prefix for a server. Also
+ // add a watcher to watch the znode.
+ try {
+ String pathPrefix = ZOOKEEPER_PATH_SEPARATOR + rootNamespace
+ + ZOOKEEPER_PATH_SEPARATOR + znodePathPrefix;
+ byte[] znodeDataUTF8 = znodeData.getBytes(StandardCharsets.UTF_8);
+ znode =
+ new PersistentEphemeralNode(zooKeeperClient,
+ PersistentEphemeralNode.Mode.EPHEMERAL_SEQUENTIAL, pathPrefix, znodeDataUTF8);
+ znode.start();
+ // We'll wait for 120s for node creation
+ long znodeCreationTimeout = 120;
+ if (!znode.waitForInitialCreate(znodeCreationTimeout, TimeUnit.SECONDS)) {
+ throw new Exception("Max znode creation wait time: " + znodeCreationTimeout + "s exhausted");
+ }
+ setDeregisteredWithZooKeeper(false);
+ final String znodePath = znode.getActualPath();
+ if ((watcher == null
+ ? zooKeeperClient.checkExists().forPath(znodePath)
+ : zooKeeperClient.checkExists().usingWatcher(watcher).forPath(znodePath))
+ == null) {
+ // No node exists, throw exception
+ throw new Exception("Unable to create znode with path prefix " + znodePathPrefix +
+ " and data " + znodeData + " on ZooKeeper.");
+ }
+ LOG.info("Created a znode (actual path " + znodePath + ") on ZooKeeper with path prefix " +
+ znodePathPrefix + " and data " + znodeData);
+ } catch (Exception e) {
+ LOG.error("Unable to create znode with path prefix " + znodePathPrefix + " and data " +
+ znodeData + " on ZooKeeper.", e);
+ if (znode != null) {
+ znode.close();
+ }
+ throw (e);
+ }
+ }
+
+ public CuratorFramework startZookeeperClient(ACLProvider zooKeeperAclProvider,
+ boolean addParentNode) throws Exception {
+ CuratorFramework zkClient = getNewZookeeperClient(zooKeeperAclProvider);
+ zkClient.start();
+
+ // Create the parent znodes recursively; ignore if the parent already exists.
+ if (addParentNode) {
+ try {
+ zkClient.create()
+ .creatingParentsIfNeeded()
+ .withMode(CreateMode.PERSISTENT)
+ .forPath(ZooKeeperHiveHelper.ZOOKEEPER_PATH_SEPARATOR + rootNamespace);
+ LOG.info("Created the root name space: " + rootNamespace + " on ZooKeeper");
+ } catch (KeeperException e) {
+ if (e.code() != KeeperException.Code.NODEEXISTS) {
+ LOG.error("Unable to create HiveServer2 namespace: " + rootNamespace + " on ZooKeeper", e);
+ throw e;
+ }
+ }
+ }
+ return zkClient;
+ }
+
+ public CuratorFramework getNewZookeeperClient() {
+ return getNewZookeeperClient(null);
+ }
+
+ public CuratorFramework getNewZookeeperClient(ACLProvider zooKeeperAclProvider) {
+ return getNewZookeeperClient(zooKeeperAclProvider, null);
+ }
+
+ public CuratorFramework getNewZookeeperClient(ACLProvider zooKeeperAclProvider,
+ String namespace) {
+ // Create a CuratorFramework instance to be used as the ZooKeeper client.
+ // Use the zooKeeperAclProvider, when specified, to create appropriate ACLs.
+ CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder()
+ .connectString(getQuorumServers())
+ .retryPolicy(new ExponentialBackoffRetry(baseSleepTime, maxRetries));
+ if (namespace != null) {
+ builder = builder.namespace(namespace);
+ }
+ if (sessionTimeout > 0) {
+ builder = builder.sessionTimeoutMs(sessionTimeout);
+ }
+ if (connectionTimeout > 0) {
+ builder = builder.connectionTimeoutMs(connectionTimeout);
+ }
+ if (sslEnabled) {
+ builder = builder.zookeeperFactory(new SSLZookeeperFactory(sslEnabled,
+ keyStoreLocation, keyStorePassword, keyStoreType,
+ trustStoreLocation, trustStorePassword, trustStoreType));
+ }
+ if (zooKeeperAclProvider != null) {
+ builder = builder.aclProvider(zooKeeperAclProvider);
+ }
+ return builder.build();
+ }
+
+ public void removeServerInstanceFromZooKeeper() throws Exception {
+ setDeregisteredWithZooKeeper(true);
+
+ if (znode != null) {
+ znode.close();
+ znode = null;
+ }
+ zooKeeperClient.close();
+ LOG.info("Server instance removed from ZooKeeper.");
+ }
+
+
+ public void deregisterZnode() {
+ if (znode != null) {
+ try {
+ znode.close();
+ LOG.warn("This server instance with path " + znode.getActualPath() +
+ " is now de-registered from ZooKeeper. ");
+ } catch (IOException e) {
+ LOG.error("Failed to close the persistent ephemeral znode", e);
+ } finally {
+ setDeregisteredWithZooKeeper(true);
+ znode = null;
+ }
+ }
+ }
+
+ public boolean isDeregisteredWithZooKeeper() {
+ return deregisteredWithZooKeeper;
+ }
+
+ private void setDeregisteredWithZooKeeper(boolean deregisteredWithZooKeeper) {
+ this.deregisteredWithZooKeeper = deregisteredWithZooKeeper;
+ }
+
+ /**
+ * This method is supposed to be called from client code connecting to one of the servers
+ * managed by the configured ZooKeeper. It starts and closes its own ZooKeeper client instead
+ * of using the class member.
+ * @return list of server URIs stored under the configured zookeeper namespace
+ * @throws Exception
+ */
+ public List getServerUris() throws Exception {
+ CuratorFramework zkClient = null;
+ List serverUris;
+ try {
+ zkClient = startZookeeperClient(null, false);
+ List serverNodes =
+ zkClient.getChildren().forPath(ZOOKEEPER_PATH_SEPARATOR + rootNamespace);
+ serverUris = new ArrayList(serverNodes.size());
+ for (String serverNode : serverNodes) {
+ byte[] serverUriBytes = zkClient.getData()
+ .forPath(ZOOKEEPER_PATH_SEPARATOR + rootNamespace +
+ ZOOKEEPER_PATH_SEPARATOR + serverNode);
+ serverUris.add(new String(serverUriBytes, StandardCharsets.UTF_8));
+ }
+ zkClient.close();
+ return serverUris;
+ } catch (Exception e) {
+ if (zkClient != null) {
+ zkClient.close();
+ }
+ throw e;
+ }
+ }
+}
diff --git a/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java b/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
index c90ad424864e..630edc86ad02 100644
--- a/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
+++ b/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
@@ -23,6 +23,7 @@
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.common.FileUtils;
+import org.apache.hadoop.hive.common.ZooKeeperHiveHelper;
import org.apache.hadoop.hive.common.classification.InterfaceAudience;
import org.apache.hadoop.hive.common.classification.InterfaceAudience.LimitedPrivate;
import org.apache.hadoop.hive.conf.Validator.PatternSet;
@@ -219,6 +220,21 @@ private static URL checkConfigFile(File f) {
HiveConf.ConfVars.REPLDIR,
HiveConf.ConfVars.METASTOREURIS,
HiveConf.ConfVars.METASTORE_SERVER_PORT,
+ HiveConf.ConfVars.METASTORE_THRIFT_BIND_HOST,
+ HiveConf.ConfVars.METASTORE_SERVICE_DISCOVERY_MODE,
+ HiveConf.ConfVars.METASTORE_ZOOKEEPER_CLIENT_PORT,
+ HiveConf.ConfVars.METASTORE_ZOOKEEPER_NAMESPACE,
+ HiveConf.ConfVars.METASTORE_ZOOKEEPER_SESSION_TIMEOUT,
+ HiveConf.ConfVars.METASTORE_ZOOKEEPER_CONNECTION_TIMEOUT,
+ HiveConf.ConfVars.METASTORE_ZOOKEEPER_CONNECTION_MAX_RETRIES,
+ HiveConf.ConfVars.METASTORE_ZOOKEEPER_CONNECTION_BASESLEEPTIME,
+ HiveConf.ConfVars.METASTORE_ZOOKEEPER_SSL_ENABLE,
+ HiveConf.ConfVars.METASTORE_ZOOKEEPER_SSL_KEYSTORE_LOCATION,
+ HiveConf.ConfVars.METASTORE_ZOOKEEPER_SSL_KEYSTORE_PASSWORD,
+ HiveConf.ConfVars.METASTORE_ZOOKEEPER_SSL_KEYSTORE_TYPE,
+ HiveConf.ConfVars.METASTORE_ZOOKEEPER_SSL_TRUSTSTORE_LOCATION,
+ HiveConf.ConfVars.METASTORE_ZOOKEEPER_SSL_TRUSTSTORE_PASSWORD,
+ HiveConf.ConfVars.METASTORE_ZOOKEEPER_SSL_TRUSTSTORE_TYPE,
HiveConf.ConfVars.METASTORETHRIFTCONNECTIONRETRIES,
HiveConf.ConfVars.METASTORETHRIFTFAILURERETRIES,
HiveConf.ConfVars.METASTORE_CLIENT_CONNECT_RETRY_DELAY,
@@ -599,7 +615,58 @@ public static enum ConfVars {
METASTOREWAREHOUSE("hive.metastore.warehouse.dir", "/user/hive/warehouse",
"location of default database for the warehouse"),
METASTOREURIS("hive.metastore.uris", "",
- "Thrift URI for the remote metastore. Used by metastore client to connect to remote metastore."),
+ "Thrift URI for the remote metastore. Used by metastore client to connect to remote " +
+ "metastore. When hive.metastore.service.discovery.mode is set, these URIs identify " +
+ "the service discovery servers instead."),
+ METASTORE_THRIFT_BIND_HOST("hive.metastore.thrift.bind.host", "",
+ "Bind host on which to run the metastore thrift service."),
+ METASTORE_SERVICE_DISCOVERY_MODE("hive.metastore.service.discovery.mode", "",
+ "Specifies which dynamic service discovery method to use. Currently only zookeeper " +
+ "is supported."),
+ METASTORE_ZOOKEEPER_CLIENT_PORT("hive.metastore.zookeeper.client.port", "2181",
+ "The port of ZooKeeper servers to talk to. If the ZooKeeper quorum specified in " +
+ "hive.metastore.uris does not contain port numbers, this value is used."),
+ METASTORE_ZOOKEEPER_SESSION_TIMEOUT("hive.metastore.zookeeper.session.timeout", "120000ms",
+ new TimeValidator(TimeUnit.MILLISECONDS),
+ "ZooKeeper client's session timeout for metastore service discovery."),
+ METASTORE_ZOOKEEPER_CONNECTION_TIMEOUT("hive.metastore.zookeeper.connection.timeout", "15s",
+ new TimeValidator(TimeUnit.SECONDS),
+ "ZooKeeper client's connection timeout for metastore service discovery."),
+ METASTORE_ZOOKEEPER_NAMESPACE("hive.metastore.zookeeper.namespace", "hive_metastore",
+ "The parent node under which all ZooKeeper nodes for metastores are created."),
+ METASTORE_ZOOKEEPER_CONNECTION_MAX_RETRIES(
+ "hive.metastore.zookeeper.connection.max.retries", 3,
+ "Max number of times to retry when connecting to the ZooKeeper server for metastore " +
+ "service discovery."),
+ METASTORE_ZOOKEEPER_CONNECTION_BASESLEEPTIME(
+ "hive.metastore.zookeeper.connection.basesleeptime", "1000ms",
+ new TimeValidator(TimeUnit.MILLISECONDS),
+ "Initial amount of time to wait between retries when connecting to ZooKeeper for " +
+ "metastore service discovery."),
+ METASTORE_ZOOKEEPER_SSL_ENABLE("hive.metastore.zookeeper.ssl.client.enable", false,
+ "Set metastore service discovery client to use TLS when connecting to ZooKeeper."),
+ METASTORE_ZOOKEEPER_SSL_KEYSTORE_LOCATION(
+ "hive.metastore.zookeeper.ssl.keystore.location", "",
+ "Keystore location when using a client-side certificate with TLS connectivity to "
+ + "ZooKeeper for metastore service discovery."),
+ METASTORE_ZOOKEEPER_SSL_KEYSTORE_PASSWORD(
+ "hive.metastore.zookeeper.ssl.keystore.password", "",
+ "Keystore password when using a client-side certificate with TLS connectivity to "
+ + "ZooKeeper for metastore service discovery."),
+ METASTORE_ZOOKEEPER_SSL_KEYSTORE_TYPE("hive.metastore.zookeeper.ssl.keystore.type", "",
+ "Keystore type when using a client-side certificate with TLS connectivity to "
+ + "ZooKeeper for metastore service discovery."),
+ METASTORE_ZOOKEEPER_SSL_TRUSTSTORE_LOCATION(
+ "hive.metastore.zookeeper.ssl.truststore.location", "",
+ "Truststore location when using TLS connectivity to ZooKeeper for metastore service "
+ + "discovery."),
+ METASTORE_ZOOKEEPER_SSL_TRUSTSTORE_PASSWORD(
+ "hive.metastore.zookeeper.ssl.truststore.password", "",
+ "Truststore password when using TLS connectivity to ZooKeeper for metastore service "
+ + "discovery."),
+ METASTORE_ZOOKEEPER_SSL_TRUSTSTORE_TYPE("hive.metastore.zookeeper.ssl.truststore.type", "",
+ "Truststore type when using TLS connectivity to ZooKeeper for metastore service "
+ + "discovery."),
METASTORE_CAPABILITY_CHECK("hive.metastore.client.capability.check", true,
"Whether to check client capabilities for potentially breaking API usage."),
@@ -1811,6 +1878,9 @@ public static enum ConfVars {
new TimeValidator(TimeUnit.MILLISECONDS),
"ZooKeeper client's session timeout (in milliseconds). The client is disconnected, and as a result, all locks released, \n" +
"if a heartbeat is not sent in the timeout."),
+ HIVE_ZOOKEEPER_CONNECTION_TIMEOUT("hive.zookeeper.connection.timeout", "15s",
+ new TimeValidator(TimeUnit.MILLISECONDS),
+ "ZooKeeper client's connection timeout."),
HIVE_ZOOKEEPER_NAMESPACE("hive.zookeeper.namespace", "hive_zookeeper_namespace",
"The parent node under which all ZooKeeper nodes are created."),
HIVE_ZOOKEEPER_CLEAN_EXTRA_NODES("hive.zookeeper.clean.extra.nodes", false,
@@ -1821,6 +1891,20 @@ public static enum ConfVars {
new TimeValidator(TimeUnit.MILLISECONDS),
"Initial amount of time (in milliseconds) to wait between retries\n" +
"when connecting to the ZooKeeper server when using ExponentialBackoffRetry policy."),
+ HIVE_ZOOKEEPER_SSL_ENABLE("hive.zookeeper.ssl.client.enable", false,
+ "Set client to use TLS when connecting to ZooKeeper."),
+ HIVE_ZOOKEEPER_SSL_KEYSTORE_LOCATION("hive.zookeeper.ssl.keystore.location", "",
+ "Keystore location when using a client-side certificate with TLS connectivity to ZooKeeper."),
+ HIVE_ZOOKEEPER_SSL_KEYSTORE_PASSWORD("hive.zookeeper.ssl.keystore.password", "",
+ "Keystore password when using a client-side certificate with TLS connectivity to ZooKeeper."),
+ HIVE_ZOOKEEPER_SSL_KEYSTORE_TYPE("hive.zookeeper.ssl.keystore.type", "",
+ "Keystore type when using a client-side certificate with TLS connectivity to ZooKeeper."),
+ HIVE_ZOOKEEPER_SSL_TRUSTSTORE_LOCATION("hive.zookeeper.ssl.truststore.location", "",
+ "Truststore location when using TLS connectivity to ZooKeeper."),
+ HIVE_ZOOKEEPER_SSL_TRUSTSTORE_PASSWORD("hive.zookeeper.ssl.truststore.password", "",
+ "Truststore password when using TLS connectivity to ZooKeeper."),
+ HIVE_ZOOKEEPER_SSL_TRUSTSTORE_TYPE("hive.zookeeper.ssl.truststore.type", "",
+ "Truststore type when using TLS connectivity to ZooKeeper."),
// Transactions
HIVE_TXN_MANAGER("hive.txn.manager",
@@ -3357,7 +3441,13 @@ public static enum ConfVars {
"hive.server2.authentication.ldap.groupMembershipKey," +
"hive.server2.authentication.ldap.userMembershipKey," +
"hive.server2.authentication.ldap.groupClassKey," +
- "hive.server2.authentication.ldap.customLDAPQuery",
+ "hive.server2.authentication.ldap.customLDAPQuery," +
+ "hive.zookeeper.ssl.keystore.location," +
+ "hive.zookeeper.ssl.keystore.password," +
+ "hive.zookeeper.ssl.keystore.type," +
+ "hive.zookeeper.ssl.truststore.location," +
+ "hive.zookeeper.ssl.truststore.password," +
+ "hive.zookeeper.ssl.truststore.type",
"Comma separated list of configuration options which are immutable at runtime"),
HIVE_CONF_HIDDEN_LIST("hive.conf.hidden.list",
METASTOREPWD.varname + "," + HIVE_SERVER2_SSL_KEYSTORE_PASSWORD.varname
@@ -3368,7 +3458,17 @@ public static enum ConfVars {
+ ",fs.s3n.awsSecretAccessKey"
+ ",fs.s3a.access.key"
+ ",fs.s3a.secret.key"
- + ",fs.s3a.proxy.password",
+ + ",fs.s3a.proxy.password"
+ + ",hive.metastore.zookeeper.ssl.keystore.location"
+ + ",hive.metastore.zookeeper.ssl.keystore.password"
+ + ",hive.metastore.zookeeper.ssl.truststore.location"
+ + ",hive.metastore.zookeeper.ssl.truststore.password"
+ + ",hive.zookeeper.ssl.keystore.location"
+ + ",hive.zookeeper.ssl.keystore.password"
+ + ",hive.zookeeper.ssl.keystore.type"
+ + ",hive.zookeeper.ssl.truststore.location"
+ + ",hive.zookeeper.ssl.truststore.password"
+ + ",hive.zookeeper.ssl.truststore.type",
"Comma separated list of configuration options which should not be read by normal user like passwords"),
HIVE_CONF_INTERNAL_VARIABLE_LIST("hive.conf.internal.variable.list",
"hive.added.files.path,hive.added.jars.path,hive.added.archives.path",
@@ -4642,4 +4742,66 @@ public static String getPassword(Configuration conf, ConfVars var) throws IOExce
}
return pw == null ? var.defaultStrVal : new String(pw);
}
+
+ public ZooKeeperHiveHelper getZKConfig() {
+ String keyStorePassword = "";
+ String trustStorePassword = "";
+ if (getBoolVar(ConfVars.HIVE_ZOOKEEPER_SSL_ENABLE)) {
+ try {
+ keyStorePassword = HiveConf.getPassword(this,
+ ConfVars.HIVE_ZOOKEEPER_SSL_KEYSTORE_PASSWORD);
+ trustStorePassword = HiveConf.getPassword(this,
+ ConfVars.HIVE_ZOOKEEPER_SSL_TRUSTSTORE_PASSWORD);
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to read zookeeper configuration passwords", e);
+ }
+ }
+ return new ZooKeeperHiveHelper(getVar(HiveConf.ConfVars.HIVE_ZOOKEEPER_QUORUM),
+ getVar(HiveConf.ConfVars.HIVE_ZOOKEEPER_CLIENT_PORT),
+ getVar(HiveConf.ConfVars.HIVE_SERVER2_ZOOKEEPER_NAMESPACE),
+ (int) getTimeVar(HiveConf.ConfVars.HIVE_ZOOKEEPER_CONNECTION_TIMEOUT,
+ TimeUnit.MILLISECONDS),
+ (int) getTimeVar(HiveConf.ConfVars.HIVE_ZOOKEEPER_SESSION_TIMEOUT, TimeUnit.MILLISECONDS),
+ (int) getTimeVar(HiveConf.ConfVars.HIVE_ZOOKEEPER_CONNECTION_BASESLEEPTIME, TimeUnit.MILLISECONDS),
+ getIntVar(HiveConf.ConfVars.HIVE_ZOOKEEPER_CONNECTION_MAX_RETRIES),
+ getBoolVar(ConfVars.HIVE_ZOOKEEPER_SSL_ENABLE),
+ getVar(ConfVars.HIVE_ZOOKEEPER_SSL_KEYSTORE_LOCATION),
+ keyStorePassword,
+ getVar(ConfVars.HIVE_ZOOKEEPER_SSL_KEYSTORE_TYPE),
+ getVar(ConfVars.HIVE_ZOOKEEPER_SSL_TRUSTSTORE_LOCATION),
+ trustStorePassword,
+ getVar(ConfVars.HIVE_ZOOKEEPER_SSL_TRUSTSTORE_TYPE));
+ }
+
+ public ZooKeeperHiveHelper getMetastoreZKConfig() {
+ String keyStorePassword = "";
+ String trustStorePassword = "";
+ if (getBoolVar(ConfVars.METASTORE_ZOOKEEPER_SSL_ENABLE)) {
+ try {
+ keyStorePassword = HiveConf.getPassword(this,
+ ConfVars.METASTORE_ZOOKEEPER_SSL_KEYSTORE_PASSWORD);
+ trustStorePassword = HiveConf.getPassword(this,
+ ConfVars.METASTORE_ZOOKEEPER_SSL_TRUSTSTORE_PASSWORD);
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to read zookeeper configuration passwords", e);
+ }
+ }
+ return new ZooKeeperHiveHelper(getVar(HiveConf.ConfVars.METASTOREURIS),
+ getVar(HiveConf.ConfVars.METASTORE_ZOOKEEPER_CLIENT_PORT),
+ getVar(HiveConf.ConfVars.METASTORE_ZOOKEEPER_NAMESPACE),
+ (int) getTimeVar(HiveConf.ConfVars.METASTORE_ZOOKEEPER_CONNECTION_TIMEOUT,
+ TimeUnit.MILLISECONDS),
+ (int) getTimeVar(HiveConf.ConfVars.METASTORE_ZOOKEEPER_SESSION_TIMEOUT,
+ TimeUnit.MILLISECONDS),
+ (int) getTimeVar(HiveConf.ConfVars.METASTORE_ZOOKEEPER_CONNECTION_BASESLEEPTIME,
+ TimeUnit.MILLISECONDS),
+ getIntVar(HiveConf.ConfVars.METASTORE_ZOOKEEPER_CONNECTION_MAX_RETRIES),
+ getBoolVar(ConfVars.METASTORE_ZOOKEEPER_SSL_ENABLE),
+ getVar(ConfVars.METASTORE_ZOOKEEPER_SSL_KEYSTORE_LOCATION),
+ keyStorePassword,
+ getVar(ConfVars.METASTORE_ZOOKEEPER_SSL_KEYSTORE_TYPE),
+ getVar(ConfVars.METASTORE_ZOOKEEPER_SSL_TRUSTSTORE_LOCATION),
+ trustStorePassword,
+ getVar(ConfVars.METASTORE_ZOOKEEPER_SSL_TRUSTSTORE_TYPE));
+ }
}
diff --git a/common/src/test/org/apache/hadoop/hive/conf/TestHiveConf.java b/common/src/test/org/apache/hadoop/hive/conf/TestHiveConf.java
index fa51ef642913..16c6bd004281 100644
--- a/common/src/test/org/apache/hadoop/hive/conf/TestHiveConf.java
+++ b/common/src/test/org/apache/hadoop/hive/conf/TestHiveConf.java
@@ -19,6 +19,7 @@
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.common.ZooKeeperHiveHelper;
import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
import org.apache.hadoop.util.Shell;
import org.apache.hive.common.util.HiveTestUtils;
@@ -26,6 +27,7 @@
import org.junit.Test;
import java.io.UnsupportedEncodingException;
+import java.lang.reflect.Field;
import java.net.URLEncoder;
import java.util.concurrent.TimeUnit;
@@ -140,6 +142,27 @@ public void testHiddenConfig() throws Exception {
Assert.assertEquals("", conf2.get(HiveConf.ConfVars.HIVE_SERVER2_SSL_KEYSTORE_PASSWORD.varname));
}
+ @Test
+ public void testZookeeperSslStoreTypes() throws Exception {
+ HiveConf conf = new HiveConf();
+ conf.setBoolVar(ConfVars.HIVE_ZOOKEEPER_SSL_ENABLE, true);
+ conf.setVar(ConfVars.HIVE_ZOOKEEPER_SSL_KEYSTORE_PASSWORD, "");
+ conf.setVar(ConfVars.HIVE_ZOOKEEPER_SSL_KEYSTORE_TYPE, "PKCS12");
+ conf.setVar(ConfVars.HIVE_ZOOKEEPER_SSL_TRUSTSTORE_PASSWORD, "");
+ conf.setVar(ConfVars.HIVE_ZOOKEEPER_SSL_TRUSTSTORE_TYPE, "BCFKS");
+
+ ZooKeeperHiveHelper zkHelper = conf.getZKConfig();
+
+ Assert.assertEquals("PKCS12", getPrivateField(zkHelper, "keyStoreType"));
+ Assert.assertEquals("BCFKS", getPrivateField(zkHelper, "trustStoreType"));
+ }
+
+ private String getPrivateField(Object target, String name) throws Exception {
+ Field field = target.getClass().getDeclaredField(name);
+ field.setAccessible(true);
+ return (String) field.get(target);
+ }
+
@Test
public void testSparkConfigUpdate(){
HiveConf conf = new HiveConf();
diff --git a/hcatalog/webhcat/svr/src/main/java/org/apache/hive/hcatalog/templeton/tool/ZooKeeperStorage.java b/hcatalog/webhcat/svr/src/main/java/org/apache/hive/hcatalog/templeton/tool/ZooKeeperStorage.java
index 8d9193faceb7..9fe422d4ced9 100644
--- a/hcatalog/webhcat/svr/src/main/java/org/apache/hive/hcatalog/templeton/tool/ZooKeeperStorage.java
+++ b/hcatalog/webhcat/svr/src/main/java/org/apache/hive/hcatalog/templeton/tool/ZooKeeperStorage.java
@@ -22,11 +22,11 @@
import java.util.ArrayList;
import java.util.List;
+import org.apache.hadoop.hive.common.ZooKeeperHiveHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
-import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.hadoop.conf.Configuration;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
@@ -52,6 +52,11 @@ public class ZooKeeperStorage implements TempletonStorage {
public static final String ZK_HOSTS = "templeton.zookeeper.hosts";
public static final String ZK_SESSION_TIMEOUT
= "templeton.zookeeper.session-timeout";
+ public static final String ZK_SSL_ENABLE = "templeton.zookeeper.ssl.client.enable";
+ public static final String ZK_KEYSTORE_LOCATION = "templeton.zookeeper.keystore.location";
+ public static final String ZK_KEYSTORE_PASSWORD = "templeton.zookeeper.keystore.password";
+ public static final String ZK_TRUSTSTORE_LOCATION = "templeton.zookeeper.truststore.location";
+ public static final String ZK_TRUSTSTORE_PASSWORD = "templeton.zookeeper.truststore.password";
public static final String ENCODING = "UTF-8";
@@ -64,10 +69,9 @@ public class ZooKeeperStorage implements TempletonStorage {
*/
public static CuratorFramework zkOpen(String zkHosts, int zkSessionTimeoutMs)
throws IOException {
- //do we need to add a connection status listener? What will that do?
- ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(1000, 3);
- CuratorFramework zk = CuratorFrameworkFactory.newClient(zkHosts, zkSessionTimeoutMs,
- CuratorFrameworkFactory.builder().getConnectionTimeoutMs(), retryPolicy);
+ ZooKeeperHiveHelper zkHelper = new ZooKeeperHiveHelper(zkHosts, null, null,
+ CuratorFrameworkFactory.builder().getConnectionTimeoutMs(), zkSessionTimeoutMs, 1000, 3);
+ CuratorFramework zk = zkHelper.getNewZookeeperClient();
zk.start();
return zk;
}
@@ -78,8 +82,20 @@ public static CuratorFramework zkOpen(String zkHosts, int zkSessionTimeoutMs)
public static CuratorFramework zkOpen(Configuration conf) throws IOException {
/*the silly looking call to Builder below is to get the default value of session timeout
from Curator which itself exposes it as system property*/
- return zkOpen(conf.get(ZK_HOSTS),
- conf.getInt(ZK_SESSION_TIMEOUT, CuratorFrameworkFactory.builder().getSessionTimeoutMs()));
+ ZooKeeperHiveHelper zkHelper = new ZooKeeperHiveHelper(conf.get(ZK_HOSTS), null, null,
+ CuratorFrameworkFactory.builder().getConnectionTimeoutMs(),
+ conf.getInt(ZK_SESSION_TIMEOUT, CuratorFrameworkFactory.builder().getSessionTimeoutMs()),
+ 1000, 3,
+ conf.getBoolean(ZK_SSL_ENABLE, false),
+ conf.get(ZK_KEYSTORE_LOCATION, ""),
+ conf.get(ZK_KEYSTORE_PASSWORD, ""),
+ null,
+ conf.get(ZK_TRUSTSTORE_LOCATION, ""),
+ conf.get(ZK_TRUSTSTORE_PASSWORD, ""),
+ null);
+ CuratorFramework zk = zkHelper.getNewZookeeperClient();
+ zk.start();
+ return zk;
}
public ZooKeeperStorage() {
diff --git a/itests/hive-unit/src/test/java/org/apache/hive/jdbc/TestServiceDiscovery.java b/itests/hive-unit/src/test/java/org/apache/hive/jdbc/TestServiceDiscovery.java
index 3aeff8937033..ceb1a3c62a2e 100644
--- a/itests/hive-unit/src/test/java/org/apache/hive/jdbc/TestServiceDiscovery.java
+++ b/itests/hive-unit/src/test/java/org/apache/hive/jdbc/TestServiceDiscovery.java
@@ -26,7 +26,7 @@
import org.apache.curator.retry.RetryOneTime;
import org.apache.curator.test.TestingServer;
import org.apache.hadoop.hive.conf.HiveConf;
-import org.apache.hadoop.hive.ql.util.ZooKeeperHiveHelper;
+import org.apache.hadoop.hive.common.ZooKeeperHiveHelper;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.junit.After;
diff --git a/itests/util/src/main/java/org/apache/hive/jdbc/miniHS2/MiniHS2.java b/itests/util/src/main/java/org/apache/hive/jdbc/miniHS2/MiniHS2.java
index a575dac5b109..04f962519a79 100644
--- a/itests/util/src/main/java/org/apache/hive/jdbc/miniHS2/MiniHS2.java
+++ b/itests/util/src/main/java/org/apache/hive/jdbc/miniHS2/MiniHS2.java
@@ -37,7 +37,6 @@
import org.apache.hadoop.hive.llap.daemon.MiniLlapCluster;
import org.apache.hadoop.hive.metastore.MetaStoreUtils;
import org.apache.hadoop.hive.ql.exec.Utilities;
-import org.apache.hadoop.hive.ql.util.ZooKeeperHiveHelper;
import org.apache.hadoop.hive.shims.HadoopShims.MiniDFSShim;
import org.apache.hadoop.hive.shims.HadoopShims.MiniMrShim;
import org.apache.hadoop.hive.shims.ShimLoader;
@@ -498,7 +497,7 @@ public String getBaseJdbcURL() {
private String getZKBaseJdbcURL() throws Exception {
HiveConf hiveConf = getServerConf();
if (hiveConf != null) {
- String zkEnsemble = ZooKeeperHiveHelper.getQuorumServers(hiveConf);
+ String zkEnsemble = hiveConf.getZKConfig().getQuorumServers();
return "jdbc:hive2://" + zkEnsemble + "/";
}
throw new Exception("Server's HiveConf is null. Unable to read ZooKeeper configs.");
diff --git a/jdbc/pom.xml b/jdbc/pom.xml
index f2cbd9f6f89b..748fc5147eb5 100644
--- a/jdbc/pom.xml
+++ b/jdbc/pom.xml
@@ -231,7 +231,8 @@
com.thoughtworks.paranamer:*
com.twitter:*
com.zaxxer:*
- io.netty:*
+ io.netty:netty
+ io.netty:netty-all
javax.activation:*
javax.inject:*
javax.jdo:*
@@ -297,6 +298,10 @@
org.apache.zookeeper
org.apache.hive.org.apache.zookeeper
+
+ io.netty
+ org.apache.hive.io.netty
+
org.apache.curator
org.apache.hive.org.apache.curator
@@ -322,6 +327,46 @@
+
+ org.apache.maven.plugins
+ maven-antrun-plugin
+
+
+ check-standalone-zookeeper-ssl-driver
+ verify
+
+ run
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/jdbc/src/java/org/apache/hive/jdbc/Utils.java b/jdbc/src/java/org/apache/hive/jdbc/Utils.java
index bfae8b9e41e4..0be2b29ea711 100644
--- a/jdbc/src/java/org/apache/hive/jdbc/Utils.java
+++ b/jdbc/src/java/org/apache/hive/jdbc/Utils.java
@@ -108,6 +108,13 @@ public static class JdbcConnectionParams {
// Use ZooKeeper for indirection while using dynamic service discovery
static final String SERVICE_DISCOVERY_MODE_ZOOKEEPER = "zooKeeper";
static final String ZOOKEEPER_NAMESPACE = "zooKeeperNamespace";
+ public static final String ZOOKEEPER_SSL_ENABLE = "zooKeeperSSLEnable";
+ public static final String ZOOKEEPER_KEYSTORE_LOCATION = "zooKeeperKeystoreLocation";
+ public static final String ZOOKEEPER_KEYSTORE_PASSWORD = "zooKeeperKeystorePassword";
+ public static final String ZOOKEEPER_KEYSTORE_TYPE = "zooKeeperKeystoreType";
+ public static final String ZOOKEEPER_TRUSTSTORE_LOCATION = "zooKeeperTruststoreLocation";
+ public static final String ZOOKEEPER_TRUSTSTORE_PASSWORD = "zooKeeperTruststorePassword";
+ public static final String ZOOKEEPER_TRUSTSTORE_TYPE = "zooKeeperTruststoreType";
// Default namespace value on ZooKeeper.
// This value is used if the param "zooKeeperNamespace" is not specified in the JDBC Uri.
static final String ZOOKEEPER_DEFAULT_NAMESPACE = "hiveserver2";
@@ -149,6 +156,13 @@ public static class JdbcConnectionParams {
private boolean isEmbeddedMode = false;
private String[] authorityList;
private String zooKeeperEnsemble = null;
+ private boolean zooKeeperSslEnabled = false;
+ private String zookeeperKeyStoreLocation = "";
+ private String zookeeperKeyStorePassword = "";
+ private String zookeeperKeyStoreType = "";
+ private String zookeeperTrustStoreLocation = "";
+ private String zookeeperTrustStorePassword = "";
+ private String zookeeperTrustStoreType = "";
private String currentHostZnodePath;
private final List rejectedHostZnodePaths = new ArrayList();
@@ -195,6 +209,34 @@ public String getZooKeeperEnsemble() {
return zooKeeperEnsemble;
}
+ public boolean isZooKeeperSslEnabled() {
+ return zooKeeperSslEnabled;
+ }
+
+ public String getZookeeperKeyStoreLocation() {
+ return zookeeperKeyStoreLocation;
+ }
+
+ public String getZookeeperKeyStorePassword() {
+ return zookeeperKeyStorePassword;
+ }
+
+ public String getZookeeperKeyStoreType() {
+ return zookeeperKeyStoreType;
+ }
+
+ public String getZookeeperTrustStoreLocation() {
+ return zookeeperTrustStoreLocation;
+ }
+
+ public String getZookeeperTrustStorePassword() {
+ return zookeeperTrustStorePassword;
+ }
+
+ public String getZookeeperTrustStoreType() {
+ return zookeeperTrustStoreType;
+ }
+
public List getRejectedHostZnodePaths() {
return rejectedHostZnodePaths;
}
@@ -243,6 +285,34 @@ public void setZooKeeperEnsemble(String zooKeeperEnsemble) {
this.zooKeeperEnsemble = zooKeeperEnsemble;
}
+ public void setZooKeeperSslEnabled(boolean zooKeeperSslEnabled) {
+ this.zooKeeperSslEnabled = zooKeeperSslEnabled;
+ }
+
+ public void setZookeeperKeyStoreLocation(String zookeeperKeyStoreLocation) {
+ this.zookeeperKeyStoreLocation = zookeeperKeyStoreLocation;
+ }
+
+ public void setZookeeperKeyStorePassword(String zookeeperKeyStorePassword) {
+ this.zookeeperKeyStorePassword = zookeeperKeyStorePassword;
+ }
+
+ public void setZookeeperKeyStoreType(String zookeeperKeyStoreType) {
+ this.zookeeperKeyStoreType = zookeeperKeyStoreType;
+ }
+
+ public void setZookeeperTrustStoreLocation(String zookeeperTrustStoreLocation) {
+ this.zookeeperTrustStoreLocation = zookeeperTrustStoreLocation;
+ }
+
+ public void setZookeeperTrustStorePassword(String zookeeperTrustStorePassword) {
+ this.zookeeperTrustStorePassword = zookeeperTrustStorePassword;
+ }
+
+ public void setZookeeperTrustStoreType(String zookeeperTrustStoreType) {
+ this.zookeeperTrustStoreType = zookeeperTrustStoreType;
+ }
+
public void setCurrentHostZnodePath(String currentHostZnodePath) {
this.currentHostZnodePath = currentHostZnodePath;
}
@@ -517,6 +587,7 @@ private static void configureConnParams(JdbcConnectionParams connParams)
.equalsIgnoreCase(serviceDiscoveryMode))) {
// Set ZooKeeper ensemble in connParams for later use
connParams.setZooKeeperEnsemble(joinStringArray(connParams.getAuthorityList(), ","));
+ ZooKeeperHiveClientHelper.setZkSSLParams(connParams);
// Configure using ZooKeeper
ZooKeeperHiveClientHelper.configureConnParams(connParams);
} else {
diff --git a/jdbc/src/java/org/apache/hive/jdbc/ZooKeeperHiveClientHelper.java b/jdbc/src/java/org/apache/hive/jdbc/ZooKeeperHiveClientHelper.java
index 8d6003ad0610..e8b04052353e 100644
--- a/jdbc/src/java/org/apache/hive/jdbc/ZooKeeperHiveClientHelper.java
+++ b/jdbc/src/java/org/apache/hive/jdbc/ZooKeeperHiveClientHelper.java
@@ -27,6 +27,7 @@
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
+import org.apache.hadoop.hive.common.SSLZookeeperFactory;
import org.apache.hive.jdbc.Utils.JdbcConnectionParams;
import org.apache.zookeeper.Watcher;
import org.slf4j.Logger;
@@ -56,9 +57,19 @@ static void configureConnParams(JdbcConnectionParams connParams)
List serverHosts;
Random randomizer = new Random();
String serverNode;
- CuratorFramework zooKeeperClient =
- CuratorFrameworkFactory.builder().connectString(zooKeeperEnsemble)
- .retryPolicy(new ExponentialBackoffRetry(1000, 3)).build();
+ CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder()
+ .connectString(zooKeeperEnsemble)
+ .retryPolicy(new ExponentialBackoffRetry(1000, 3));
+ if (connParams.isZooKeeperSslEnabled()) {
+ builder = builder.zookeeperFactory(new SSLZookeeperFactory(true,
+ connParams.getZookeeperKeyStoreLocation(),
+ connParams.getZookeeperKeyStorePassword(),
+ connParams.getZookeeperKeyStoreType(),
+ connParams.getZookeeperTrustStoreLocation(),
+ connParams.getZookeeperTrustStorePassword(),
+ connParams.getZookeeperTrustStoreType()));
+ }
+ CuratorFramework zooKeeperClient = builder.build();
try {
zooKeeperClient.start();
serverHosts = zooKeeperClient.getChildren().forPath("/" + zooKeeperNamespace);
@@ -102,6 +113,37 @@ static void configureConnParams(JdbcConnectionParams connParams)
}
}
+ /**
+ * Parse and set up SSL communication parameters for ZooKeeper service discovery.
+ */
+ static void setZkSSLParams(JdbcConnectionParams connParams) {
+ boolean sslEnabled = false;
+ if (connParams.getSessionVars().containsKey(JdbcConnectionParams.ZOOKEEPER_SSL_ENABLE)) {
+ sslEnabled = Boolean.parseBoolean(
+ connParams.getSessionVars().get(JdbcConnectionParams.ZOOKEEPER_SSL_ENABLE));
+ connParams.setZooKeeperSslEnabled(sslEnabled);
+ }
+ if (sslEnabled) {
+ connParams.setZookeeperKeyStoreLocation(getSessionVarOrEmpty(connParams,
+ JdbcConnectionParams.ZOOKEEPER_KEYSTORE_LOCATION));
+ connParams.setZookeeperKeyStorePassword(getSessionVarOrEmpty(connParams,
+ JdbcConnectionParams.ZOOKEEPER_KEYSTORE_PASSWORD));
+ connParams.setZookeeperKeyStoreType(getSessionVarOrEmpty(connParams,
+ JdbcConnectionParams.ZOOKEEPER_KEYSTORE_TYPE));
+ connParams.setZookeeperTrustStoreLocation(getSessionVarOrEmpty(connParams,
+ JdbcConnectionParams.ZOOKEEPER_TRUSTSTORE_LOCATION));
+ connParams.setZookeeperTrustStorePassword(getSessionVarOrEmpty(connParams,
+ JdbcConnectionParams.ZOOKEEPER_TRUSTSTORE_PASSWORD));
+ connParams.setZookeeperTrustStoreType(getSessionVarOrEmpty(connParams,
+ JdbcConnectionParams.ZOOKEEPER_TRUSTSTORE_TYPE));
+ }
+ }
+
+ private static String getSessionVarOrEmpty(JdbcConnectionParams connParams, String key) {
+ String value = connParams.getSessionVars().get(key);
+ return value == null ? "" : value;
+ }
+
/**
* Apply configs published by the server. Configs specified from client's JDBC URI override
* configs published by the server.
diff --git a/jdbc/src/test/org/apache/hive/jdbc/TestZooKeeperHiveClientHelper.java b/jdbc/src/test/org/apache/hive/jdbc/TestZooKeeperHiveClientHelper.java
new file mode 100644
index 000000000000..31f1851f7394
--- /dev/null
+++ b/jdbc/src/test/org/apache/hive/jdbc/TestZooKeeperHiveClientHelper.java
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hive.jdbc;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import org.apache.hive.jdbc.Utils.JdbcConnectionParams;
+import org.junit.Test;
+
+public class TestZooKeeperHiveClientHelper {
+
+ @Test
+ public void testSetZkSSLParamsReadsStoreTypes() {
+ JdbcConnectionParams connParams = new JdbcConnectionParams();
+ connParams.getSessionVars().put(JdbcConnectionParams.ZOOKEEPER_SSL_ENABLE, "true");
+ connParams.getSessionVars().put(JdbcConnectionParams.ZOOKEEPER_KEYSTORE_LOCATION,
+ "/tmp/client.p12");
+ connParams.getSessionVars().put(JdbcConnectionParams.ZOOKEEPER_KEYSTORE_PASSWORD, "keypass");
+ connParams.getSessionVars().put(JdbcConnectionParams.ZOOKEEPER_KEYSTORE_TYPE, "PKCS12");
+ connParams.getSessionVars().put(JdbcConnectionParams.ZOOKEEPER_TRUSTSTORE_LOCATION,
+ "/tmp/truststore.bcfks");
+ connParams.getSessionVars().put(JdbcConnectionParams.ZOOKEEPER_TRUSTSTORE_PASSWORD,
+ "trustpass");
+ connParams.getSessionVars().put(JdbcConnectionParams.ZOOKEEPER_TRUSTSTORE_TYPE, "BCFKS");
+
+ ZooKeeperHiveClientHelper.setZkSSLParams(connParams);
+
+ assertTrue(connParams.isZooKeeperSslEnabled());
+ assertEquals("/tmp/client.p12", connParams.getZookeeperKeyStoreLocation());
+ assertEquals("keypass", connParams.getZookeeperKeyStorePassword());
+ assertEquals("PKCS12", connParams.getZookeeperKeyStoreType());
+ assertEquals("/tmp/truststore.bcfks", connParams.getZookeeperTrustStoreLocation());
+ assertEquals("trustpass", connParams.getZookeeperTrustStorePassword());
+ assertEquals("BCFKS", connParams.getZookeeperTrustStoreType());
+ }
+}
diff --git a/llap-client/src/java/org/apache/hadoop/hive/llap/registry/impl/LlapZookeeperRegistryImpl.java b/llap-client/src/java/org/apache/hadoop/hive/llap/registry/impl/LlapZookeeperRegistryImpl.java
index ad1714417712..2c0dab905b78 100644
--- a/llap-client/src/java/org/apache/hadoop/hive/llap/registry/impl/LlapZookeeperRegistryImpl.java
+++ b/llap-client/src/java/org/apache/hadoop/hive/llap/registry/impl/LlapZookeeperRegistryImpl.java
@@ -42,7 +42,6 @@
import com.google.common.collect.Sets;
import org.apache.curator.framework.CuratorFramework;
-import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.ACLProvider;
import org.apache.curator.framework.imps.CuratorFrameworkState;
import org.apache.curator.framework.recipes.cache.ChildData;
@@ -51,7 +50,6 @@
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.framework.recipes.nodes.PersistentEphemeralNode;
import org.apache.curator.framework.recipes.nodes.PersistentEphemeralNode.Mode;
-import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.utils.CloseableUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.conf.HiveConf;
@@ -149,14 +147,7 @@ public class LlapZookeeperRegistryImpl implements ServiceRegistry {
public LlapZookeeperRegistryImpl(String instanceName, Configuration conf) {
this.conf = new Configuration(conf);
this.conf.addResource(YarnConfiguration.YARN_SITE_CONFIGURATION_FILE);
- String zkEnsemble = getQuorumServers(this.conf);
this.encoder = new RegistryUtils.ServiceRecordMarshal();
- int sessionTimeout = (int) HiveConf.getTimeVar(conf, ConfVars.HIVE_ZOOKEEPER_SESSION_TIMEOUT,
- TimeUnit.MILLISECONDS);
- int baseSleepTime = (int) HiveConf
- .getTimeVar(conf, ConfVars.HIVE_ZOOKEEPER_CONNECTION_BASESLEEPTIME,
- TimeUnit.MILLISECONDS);
- int maxRetries = HiveConf.getIntVar(conf, ConfVars.HIVE_ZOOKEEPER_CONNECTION_MAX_RETRIES);
// sample path: /llap-sasl/hiveuser/hostname/workers/worker-0000000
// worker-0000000 is the sequence number which will be retained until session timeout. If a
@@ -194,15 +185,9 @@ public List getAclForPath(String path) {
rootNs = isSecure ? SASL_NAMESPACE : UNSECURE_NAMESPACE; // The normal path.
}
- // Create a CuratorFramework instance to be used as the ZooKeeper client
- // Use the zooKeeperAclProvider to create appropriate ACLs
- this.zooKeeperClient = CuratorFrameworkFactory.builder()
- .connectString(zkEnsemble)
- .sessionTimeoutMs(sessionTimeout)
- .aclProvider(zooKeeperAclProvider)
- .namespace(rootNs)
- .retryPolicy(new ExponentialBackoffRetry(baseSleepTime, maxRetries))
- .build();
+ HiveConf zkConf = new HiveConf(this.conf, LlapZookeeperRegistryImpl.class);
+ this.zooKeeperClient = zkConf.getZKConfig().getNewZookeeperClient(
+ zooKeeperAclProvider, rootNs);
LOG.info("Llap Zookeeper Registry is enabled with registryid: " + instanceName);
}
diff --git a/metastore/pom.xml b/metastore/pom.xml
index b229d7b2ec5c..aa97c2279a6c 100644
--- a/metastore/pom.xml
+++ b/metastore/pom.xml
@@ -243,6 +243,12 @@
${junit.version}
test
+
+ org.apache.curator
+ curator-test
+ ${curator.version}
+ test
+
org.mockito
mockito-all
diff --git a/metastore/src/java/org/apache/hadoop/hive/metastore/HiveMetaStore.java b/metastore/src/java/org/apache/hadoop/hive/metastore/HiveMetaStore.java
index 0ca5992518de..0bd5fcd5f1c8 100644
--- a/metastore/src/java/org/apache/hadoop/hive/metastore/HiveMetaStore.java
+++ b/metastore/src/java/org/apache/hadoop/hive/metastore/HiveMetaStore.java
@@ -24,6 +24,7 @@
import static org.apache.hadoop.hive.metastore.MetaStoreUtils.validateName;
import java.io.IOException;
+import java.net.InetAddress;
import java.nio.ByteBuffer;
import java.security.PrivilegedExceptionAction;
import java.text.DateFormat;
@@ -70,6 +71,7 @@
import org.apache.hadoop.hive.common.LogUtils;
import org.apache.hadoop.hive.common.LogUtils.LogInitializationException;
import org.apache.hadoop.hive.common.StatsSetupConst;
+import org.apache.hadoop.hive.common.ZooKeeperHiveHelper;
import org.apache.hadoop.hive.common.auth.HiveAuthUtils;
import org.apache.hadoop.hive.common.classification.InterfaceAudience;
import org.apache.hadoop.hive.common.classification.InterfaceStability;
@@ -198,6 +200,8 @@ protected DateFormat initialValue() {
public static final String NO_FILTER_STRING = "";
public static final int UNLIMITED_MAX_PARTITIONS = -1;
+ private static ZooKeeperHiveHelper zooKeeperHelper = null;
+ private static String metastoreBindHost = null;
private static final class ChainedTTransportFactory extends TTransportFactory {
private final TTransportFactory parentTransFactory;
@@ -7056,6 +7060,14 @@ public void run() {
+ e.getMessage(), e);
}
}
+ if (conf.getVar(ConfVars.METASTORE_SERVICE_DISCOVERY_MODE).equalsIgnoreCase("zookeeper")
+ && zooKeeperHelper != null) {
+ try {
+ zooKeeperHelper.removeServerInstanceFromZooKeeper();
+ } catch (Exception e) {
+ LOG.error("Error removing znode for this metastore instance from ZooKeeper.", e);
+ }
+ }
}
});
@@ -7149,6 +7161,13 @@ public static void startMetaStore(int port, HadoopThriftAuthBridge bridge,
false);
IHMSHandler handler = newRetryingHMSHandler(baseHandler, conf);
TServerSocket serverSocket = null;
+ metastoreBindHost = conf.getVar(ConfVars.METASTORE_THRIFT_BIND_HOST);
+ if (metastoreBindHost != null && metastoreBindHost.trim().isEmpty()) {
+ metastoreBindHost = null;
+ }
+ if (metastoreBindHost != null) {
+ LOG.info("Binding metastore thrift service to host " + metastoreBindHost);
+ }
if (useSasl) {
// we are in secure mode.
@@ -7167,7 +7186,7 @@ public static void startMetaStore(int port, HadoopThriftAuthBridge bridge,
MetaStoreUtils.getMetaStoreSaslProperties(conf));
processor = saslServer.wrapProcessor(
new ThriftHiveMetastore.Processor(handler));
- serverSocket = HiveAuthUtils.getServerSocket(null, port);
+ serverSocket = HiveAuthUtils.getServerSocket(metastoreBindHost, port);
LOG.info("Starting DB backed MetaStore Server in Secure Mode");
} else {
@@ -7193,7 +7212,7 @@ public static void startMetaStore(int port, HadoopThriftAuthBridge bridge,
sslVersionBlacklist.add(sslVersion);
}
if (!useSSL) {
- serverSocket = HiveAuthUtils.getServerSocket(null, port);
+ serverSocket = HiveAuthUtils.getServerSocket(metastoreBindHost, port);
} else {
String keyStorePath = conf.getVar(ConfVars.HIVE_METASTORE_SSL_KEYSTORE_PATH).trim();
if (keyStorePath.isEmpty()) {
@@ -7202,7 +7221,7 @@ public static void startMetaStore(int port, HadoopThriftAuthBridge bridge,
}
String keyStorePassword = ShimLoader.getHadoopShims().getPassword(conf,
HiveConf.ConfVars.HIVE_METASTORE_SSL_KEYSTORE_PASSWORD.varname);
- serverSocket = HiveAuthUtils.getServerSSLSocket(null, port, keyStorePath,
+ serverSocket = HiveAuthUtils.getServerSSLSocket(metastoreBindHost, port, keyStorePath,
keyStorePassword, sslVersionBlacklist);
}
}
@@ -7270,6 +7289,19 @@ public void processContext(ServerContext serverContext, TTransport tTransport, T
if (startLock != null) {
signalOtherThreadsToStart(tServer, startLock, startCondition, startedServing);
}
+ if (conf.getVar(ConfVars.METASTORE_SERVICE_DISCOVERY_MODE).equalsIgnoreCase("zookeeper")) {
+ try {
+ zooKeeperHelper = conf.getMetastoreZKConfig();
+ String serverInstanceURI = getServerInstanceURI(port);
+ zooKeeperHelper.addServerInstanceToZooKeeper(serverInstanceURI, serverInstanceURI, null,
+ null);
+ HMSHandler.LOG.info("Metastore server instance with URL " + serverInstanceURI
+ + " added to ZooKeeper.");
+ } catch (Exception e) {
+ LOG.error("Error adding this metastore instance to ZooKeeper: ", e);
+ throw e;
+ }
+ }
tServer.serve();
} catch (Throwable x) {
x.printStackTrace();
@@ -7278,6 +7310,14 @@ public void processContext(ServerContext serverContext, TTransport tTransport, T
}
}
+ private static String getServerInstanceURI(int port) throws Exception {
+ String hostName = metastoreBindHost;
+ if (hostName == null || hostName.trim().isEmpty()) {
+ hostName = InetAddress.getLocalHost().getHostName();
+ }
+ return hostName + ":" + port;
+ }
+
private static void cleanupRawStore() {
RawStore rs = HMSHandler.getRawStore();
if (rs != null) {
diff --git a/metastore/src/java/org/apache/hadoop/hive/metastore/HiveMetaStoreClient.java b/metastore/src/java/org/apache/hadoop/hive/metastore/HiveMetaStoreClient.java
index 394580b702e1..8e129a8cf17a 100644
--- a/metastore/src/java/org/apache/hadoop/hive/metastore/HiveMetaStoreClient.java
+++ b/metastore/src/java/org/apache/hadoop/hive/metastore/HiveMetaStoreClient.java
@@ -154,7 +154,9 @@ public HiveMetaStoreClient(HiveConf conf, HiveMetaHookLoader hookLoader, Boolean
conf, HiveConf.ConfVars.METASTORE_BATCH_RETRIEVE_OBJECTS_MAX);
String msUri = conf.getVar(ConfVars.METASTOREURIS);
- localMetaStore = HiveConfUtil.isEmbeddedMetaStore(msUri);
+ String serviceDiscoveryMode = conf.getVar(ConfVars.METASTORE_SERVICE_DISCOVERY_MODE);
+ localMetaStore = HiveConfUtil.isEmbeddedMetaStore(msUri)
+ && (serviceDiscoveryMode == null || serviceDiscoveryMode.trim().isEmpty());
if (localMetaStore) {
if (!allowEmbedded) {
throw new MetaException("Embedded metastore is not allowed here. Please configure "
@@ -183,35 +185,7 @@ public HiveMetaStoreClient(HiveConf conf, HiveMetaHookLoader hookLoader, Boolean
retryDelaySeconds = conf.getTimeVar(
ConfVars.METASTORE_CLIENT_CONNECT_RETRY_DELAY, TimeUnit.SECONDS);
- // user wants file store based configuration
- if (conf.getVar(HiveConf.ConfVars.METASTOREURIS) != null) {
- String metastoreUrisString[] = conf.getVar(
- HiveConf.ConfVars.METASTOREURIS).split(",");
- metastoreUris = new URI[metastoreUrisString.length];
- try {
- int i = 0;
- for (String s : metastoreUrisString) {
- URI tmpUri = new URI(s);
- if (tmpUri.getScheme() == null) {
- throw new IllegalArgumentException("URI: " + s
- + " does not have a scheme");
- }
- metastoreUris[i++] = tmpUri;
-
- }
- // make metastore URIS random
- List uriList = Arrays.asList(metastoreUris);
- Collections.shuffle(uriList);
- metastoreUris = uriList.toArray(new URI[uriList.size()]);
- } catch (IllegalArgumentException e) {
- throw (e);
- } catch (Exception e) {
- MetaStoreUtils.logAndThrowMetaException(e);
- }
- } else {
- LOG.error("NOT getting uris from conf");
- throw new MetaException("MetaStoreURIs not found in conf file");
- }
+ resolveMetastoreUris();
//If HADOOP_PROXY_USER is set in env or property,
//then need to create metastore client that proxies as that user.
@@ -252,6 +226,59 @@ public Void run() throws Exception {
open();
}
+ private void resolveMetastoreUris() throws MetaException {
+ String metastoreUrisConf = conf.getVar(HiveConf.ConfVars.METASTOREURIS);
+ if (metastoreUrisConf == null) {
+ LOG.error("NOT getting uris from conf");
+ throw new MetaException("MetaStoreURIs not found in conf file");
+ }
+
+ String serviceDiscoveryMode = conf.getVar(ConfVars.METASTORE_SERVICE_DISCOVERY_MODE);
+ List metastoreUrisString = new ArrayList<>();
+ try {
+ if (serviceDiscoveryMode == null || serviceDiscoveryMode.trim().isEmpty()) {
+ metastoreUrisString.addAll(Arrays.asList(metastoreUrisConf.split(",")));
+ } else if (serviceDiscoveryMode.equalsIgnoreCase("zookeeper")) {
+ for (String serverUri : conf.getMetastoreZKConfig().getServerUris()) {
+ metastoreUrisString.add("thrift://" + serverUri);
+ }
+ } else {
+ throw new IllegalArgumentException("Invalid metastore dynamic service discovery mode "
+ + serviceDiscoveryMode);
+ }
+ if (metastoreUrisString.isEmpty()
+ && "zookeeper".equalsIgnoreCase(serviceDiscoveryMode)) {
+ throw new MetaException("No metastore service discovered in ZooKeeper. "
+ + "Please ensure that at least one metastore server is online");
+ }
+
+ URI[] resolvedMetastoreUris = new URI[metastoreUrisString.size()];
+ int i = 0;
+ for (String s : metastoreUrisString) {
+ URI tmpUri = new URI(s);
+ if (tmpUri.getScheme() == null) {
+ throw new IllegalArgumentException("URI: " + s
+ + " does not have a scheme");
+ }
+ resolvedMetastoreUris[i++] = tmpUri;
+ }
+
+ // make metastore URIS random
+ List uriList = Arrays.asList(resolvedMetastoreUris);
+ Collections.shuffle(uriList);
+ metastoreUris = uriList.toArray(new URI[uriList.size()]);
+ } catch (IllegalArgumentException e) {
+ throw (e);
+ } catch (Exception e) {
+ MetaStoreUtils.logAndThrowMetaException(e);
+ }
+ }
+
+ private boolean isZooKeeperServiceDiscoveryEnabled() {
+ String serviceDiscoveryMode = conf.getVar(ConfVars.METASTORE_SERVICE_DISCOVERY_MODE);
+ return "zookeeper".equalsIgnoreCase(serviceDiscoveryMode);
+ }
+
private MetaStoreFilterHook loadFilterHooks() throws IllegalStateException {
Class authProviderClass = conf.
getClass(HiveConf.ConfVars.METASTORE_FILTER_HOOK.varname,
@@ -338,6 +365,9 @@ public void reconnect() throws MetaException {
" at the client level.");
} else {
close();
+ if (isZooKeeperServiceDiscoveryEnabled()) {
+ resolveMetastoreUris();
+ }
// Swap the first element of the metastoreUris[] with a random element from the rest
// of the array. Rationale being that this method will generally be called when the default
// connection has died and the default connection is likely to be the first array element.
diff --git a/metastore/src/java/org/apache/hadoop/hive/metastore/MetaStoreUtils.java b/metastore/src/java/org/apache/hadoop/hive/metastore/MetaStoreUtils.java
index e62bcc5c584d..f9377ab3228b 100644
--- a/metastore/src/java/org/apache/hadoop/hive/metastore/MetaStoreUtils.java
+++ b/metastore/src/java/org/apache/hadoop/hive/metastore/MetaStoreUtils.java
@@ -26,6 +26,7 @@
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
+import java.net.SocketAddress;
import java.net.URL;
import java.net.URLClassLoader;
import java.util.ArrayList;
@@ -1281,6 +1282,10 @@ public static int startMetaStoreWithRetry(final HadoopThriftAuthBridge bridge, H
for (int tryCount = 0; tryCount < MetaStoreUtils.RETRY_COUNT; tryCount++) {
try {
metaStorePort = findFreePort();
+ if (conf != null && conf.getVar(HiveConf.ConfVars.METASTORE_SERVICE_DISCOVERY_MODE)
+ .trim().isEmpty()) {
+ conf.setVar(HiveConf.ConfVars.METASTOREURIS, "thrift://localhost:" + metaStorePort);
+ }
startMetaStore(metaStorePort, bridge, conf);
return metaStorePort;
} catch (ConnectException ce) {
@@ -1318,20 +1323,23 @@ public void run() {
});
thread.setDaemon(true);
thread.start();
- loopUntilHMSReady(port);
+ loopUntilHMSReady(hiveConf.getVar(HiveConf.ConfVars.METASTORE_THRIFT_BIND_HOST), port);
}
/**
* A simple connect test to make sure that the metastore is up
* @throws Exception
*/
- private static void loopUntilHMSReady(int port) throws Exception {
+ private static void loopUntilHMSReady(String msHost, int port) throws Exception {
int retries = 0;
Exception exc = null;
while (true) {
try {
Socket socket = new Socket();
- socket.connect(new InetSocketAddress(port), 5000);
+ SocketAddress sockAddr = (msHost == null || msHost.trim().isEmpty())
+ ? new InetSocketAddress(port)
+ : new InetSocketAddress(msHost, port);
+ socket.connect(sockAddr, 5000);
socket.close();
return;
} catch (Exception e) {
diff --git a/metastore/src/test/org/apache/hadoop/hive/metastore/TestRemoteHMSZKNegative.java b/metastore/src/test/org/apache/hadoop/hive/metastore/TestRemoteHMSZKNegative.java
new file mode 100644
index 000000000000..9cb795650980
--- /dev/null
+++ b/metastore/src/test/org/apache/hadoop/hive/metastore/TestRemoteHMSZKNegative.java
@@ -0,0 +1,82 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hive.metastore;
+
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.retry.RetryOneTime;
+import org.apache.curator.test.TestingServer;
+import org.apache.hadoop.hive.common.ZooKeeperHiveHelper;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
+import org.apache.hadoop.hive.metastore.api.MetaException;
+import org.apache.zookeeper.CreateMode;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+public class TestRemoteHMSZKNegative {
+ private TestingServer zkServer;
+ private CuratorFramework zkClient;
+ private HiveConf conf;
+
+ @Before
+ public void setUp() throws Exception {
+ String rootNamespace = getClass().getSimpleName();
+ zkServer = new TestingServer();
+
+ conf = new HiveConf(TestRemoteHMSZKNegative.class);
+ conf.setVar(ConfVars.METASTOREURIS, zkServer.getConnectString());
+ conf.setVar(ConfVars.METASTORE_ZOOKEEPER_NAMESPACE, rootNamespace);
+ conf.setVar(ConfVars.METASTORE_SERVICE_DISCOVERY_MODE, "zookeeper");
+
+ zkClient = CuratorFrameworkFactory.newClient(zkServer.getConnectString(),
+ new RetryOneTime(2000));
+ zkClient.start();
+ zkClient.create()
+ .creatingParentsIfNeeded()
+ .withMode(CreateMode.PERSISTENT)
+ .forPath(ZooKeeperHiveHelper.ZOOKEEPER_PATH_SEPARATOR + rootNamespace);
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ if (zkClient != null) {
+ zkClient.close();
+ zkClient = null;
+ }
+ if (zkServer != null) {
+ zkServer.close();
+ zkServer = null;
+ }
+ }
+
+ @Test
+ public void testClientThrowsWhenNoMetaStoreRegisteredInZooKeeper() {
+ try {
+ new HiveMetaStoreClient(conf);
+ fail("Expected MetaException");
+ } catch (Exception e) {
+ assertTrue(e instanceof MetaException);
+ assertTrue(e.getMessage().contains("No metastore service discovered in ZooKeeper"));
+ }
+ }
+}
diff --git a/metastore/src/test/org/apache/hadoop/hive/metastore/TestRemoteHiveMetaStoreZK.java b/metastore/src/test/org/apache/hadoop/hive/metastore/TestRemoteHiveMetaStoreZK.java
new file mode 100644
index 000000000000..41cabadef2d4
--- /dev/null
+++ b/metastore/src/test/org/apache/hadoop/hive/metastore/TestRemoteHiveMetaStoreZK.java
@@ -0,0 +1,198 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hive.metastore;
+
+import java.lang.reflect.Field;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.curator.test.TestingServer;
+import org.apache.hadoop.hive.common.ZooKeeperHiveHelper;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
+import org.apache.hadoop.hive.metastore.api.Database;
+import org.apache.hadoop.hive.metastore.api.NoSuchObjectException;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+
+public class TestRemoteHiveMetaStoreZK {
+ private static TestingServer zkServer;
+ private static HiveConf hiveConf;
+ private static HiveMetaStoreClient client;
+
+ @BeforeClass
+ public static void startMetaStoreServer() throws Exception {
+ zkServer = new TestingServer();
+
+ HiveConf metastoreConf = new HiveConf(TestRemoteHiveMetaStoreZK.class);
+ configureMetaStore(metastoreConf);
+ configureZooKeeperServiceDiscovery(metastoreConf);
+ MetaStoreUtils.startMetaStoreWithRetry(metastoreConf);
+ waitForMetaStoreRegistration(metastoreConf);
+
+ hiveConf = new HiveConf(TestRemoteHiveMetaStoreZK.class);
+ configureMetaStore(hiveConf);
+ configureZooKeeperServiceDiscovery(hiveConf);
+ }
+
+ @AfterClass
+ public static void stopZooKeeperServer() throws Exception {
+ if (zkServer != null) {
+ zkServer.close();
+ zkServer = null;
+ }
+ }
+
+ @Before
+ public void createClient() throws Exception {
+ client = new HiveMetaStoreClient(hiveConf);
+ }
+
+ @After
+ public void closeClient() {
+ if (client != null) {
+ client.close();
+ client = null;
+ }
+ }
+
+ @Test
+ public void testClientConnectsThroughZooKeeper() throws Exception {
+ String dbName = "test_remote_hms_zk";
+ Database db = new Database();
+ db.setName(dbName);
+
+ dropDatabaseIfExists(dbName);
+ client.createDatabase(db);
+
+ try {
+ assertEquals(dbName, client.getDatabase(dbName).getName());
+ } finally {
+ client.dropDatabase(dbName, true, true, true);
+ }
+ }
+
+ @Test
+ public void testClientConnectsAfterMetaStoreRegistersAgain() throws Exception {
+ closeClient();
+ removeMetaStoreRegistration();
+ waitForNoMetaStoreRegistration(hiveConf);
+
+ HiveConf restartedMetaStoreConf = new HiveConf(TestRemoteHiveMetaStoreZK.class);
+ configureMetaStore(restartedMetaStoreConf);
+ configureZooKeeperServiceDiscovery(restartedMetaStoreConf);
+ MetaStoreUtils.startMetaStoreWithRetry(restartedMetaStoreConf);
+ waitForMetaStoreRegistration(restartedMetaStoreConf);
+
+ client = new HiveMetaStoreClient(hiveConf);
+ assertEquals("default", client.getDatabase("default").getName());
+ }
+
+ @Test
+ public void testClientConnectsAfterZooKeeperRestart() throws Exception {
+ closeClient();
+
+ zkServer.stop();
+ zkServer.restart();
+ waitForMetaStoreRegistration(hiveConf);
+
+ client = new HiveMetaStoreClient(hiveConf);
+ assertEquals("default", client.getDatabase("default").getName());
+ }
+
+ protected static void configureZooKeeperServiceDiscovery(HiveConf conf) {
+ conf.setVar(ConfVars.METASTOREURIS, zkServer.getConnectString());
+ conf.setVar(ConfVars.METASTORE_ZOOKEEPER_NAMESPACE,
+ TestRemoteHiveMetaStoreZK.class.getSimpleName());
+ conf.setVar(ConfVars.METASTORE_SERVICE_DISCOVERY_MODE, "zookeeper");
+ }
+
+ private static void configureMetaStore(HiveConf conf) {
+ conf.setClass(ConfVars.METASTORE_EXPRESSION_PROXY_CLASS.varname,
+ MockPartitionExpressionForMetastore.class, PartitionExpressionProxy.class);
+ conf.setIntVar(ConfVars.METASTORETHRIFTCONNECTIONRETRIES, 3);
+ conf.setTimeVar(ConfVars.METASTORE_CLIENT_CONNECT_RETRY_DELAY, 100,
+ TimeUnit.MILLISECONDS);
+ conf.setTimeVar(ConfVars.METASTORE_CLIENT_CONNECTION_TIMEOUT, 5,
+ TimeUnit.SECONDS);
+ conf.setTimeVar(ConfVars.METASTORE_CLIENT_SOCKET_TIMEOUT, 30,
+ TimeUnit.SECONDS);
+ }
+
+ private static void waitForMetaStoreRegistration(HiveConf conf) throws Exception {
+ long deadline = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(30);
+ Exception lastException = null;
+ while (System.currentTimeMillis() < deadline) {
+ try {
+ List serverUris = conf.getMetastoreZKConfig().getServerUris();
+ if (!serverUris.isEmpty()) {
+ return;
+ }
+ } catch (Exception e) {
+ lastException = e;
+ }
+ Thread.sleep(250);
+ }
+
+ AssertionError error = new AssertionError("No metastore instance registered in ZooKeeper "
+ + "namespace " + conf.getVar(ConfVars.METASTORE_ZOOKEEPER_NAMESPACE));
+ if (lastException != null) {
+ error.initCause(lastException);
+ }
+ throw error;
+ }
+
+ private static void waitForNoMetaStoreRegistration(HiveConf conf) throws Exception {
+ long deadline = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(30);
+ List serverUris = null;
+ while (System.currentTimeMillis() < deadline) {
+ serverUris = conf.getMetastoreZKConfig().getServerUris();
+ if (serverUris.isEmpty()) {
+ return;
+ }
+ Thread.sleep(250);
+ }
+
+ throw new AssertionError("Metastore instances are still registered in ZooKeeper namespace "
+ + conf.getVar(ConfVars.METASTORE_ZOOKEEPER_NAMESPACE) + ": " + serverUris);
+ }
+
+ private static void removeMetaStoreRegistration() throws Exception {
+ ZooKeeperHiveHelper zooKeeperHiveHelper = getMetaStoreZooKeeperHelper();
+ zooKeeperHiveHelper.removeServerInstanceFromZooKeeper();
+ }
+
+ private static ZooKeeperHiveHelper getMetaStoreZooKeeperHelper() throws Exception {
+ Field field = HiveMetaStore.class.getDeclaredField("zooKeeperHelper");
+ field.setAccessible(true);
+ return (ZooKeeperHiveHelper) field.get(null);
+ }
+
+ private static void dropDatabaseIfExists(String dbName) throws Exception {
+ try {
+ client.dropDatabase(dbName, true, true, true);
+ } catch (NoSuchObjectException e) {
+ // Ignore missing databases left from previous cleanup attempts.
+ }
+ }
+}
diff --git a/metastore/src/test/org/apache/hadoop/hive/metastore/TestRemoteHiveMetaStoreZKBindHost.java b/metastore/src/test/org/apache/hadoop/hive/metastore/TestRemoteHiveMetaStoreZKBindHost.java
new file mode 100644
index 000000000000..7dcc02b24af0
--- /dev/null
+++ b/metastore/src/test/org/apache/hadoop/hive/metastore/TestRemoteHiveMetaStoreZKBindHost.java
@@ -0,0 +1,143 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hive.metastore;
+
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.curator.test.TestingServer;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
+import org.apache.hadoop.hive.metastore.api.Database;
+import org.apache.hadoop.hive.metastore.api.NoSuchObjectException;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+
+public class TestRemoteHiveMetaStoreZKBindHost {
+ private static TestingServer zkServer;
+ private static HiveConf hiveConf;
+ private static HiveMetaStoreClient client;
+
+ @BeforeClass
+ public static void startMetaStoreServer() throws Exception {
+ zkServer = new TestingServer();
+
+ HiveConf metastoreConf = new HiveConf(TestRemoteHiveMetaStoreZKBindHost.class);
+ configureMetaStore(metastoreConf);
+ configureZooKeeperServiceDiscovery(metastoreConf);
+ MetaStoreUtils.startMetaStoreWithRetry(metastoreConf);
+ waitForMetaStoreRegistration(metastoreConf);
+
+ hiveConf = new HiveConf(TestRemoteHiveMetaStoreZKBindHost.class);
+ configureMetaStore(hiveConf);
+ configureZooKeeperServiceDiscovery(hiveConf);
+ }
+
+ @AfterClass
+ public static void stopZooKeeperServer() throws Exception {
+ if (zkServer != null) {
+ zkServer.close();
+ zkServer = null;
+ }
+ }
+
+ @Before
+ public void createClient() throws Exception {
+ client = new HiveMetaStoreClient(hiveConf);
+ }
+
+ @After
+ public void closeClient() {
+ if (client != null) {
+ client.close();
+ client = null;
+ }
+ }
+
+ @Test
+ public void testClientConnectsThroughZooKeeperWithBindHost() throws Exception {
+ String dbName = "test_remote_hms_zk_bind_host";
+ Database db = new Database();
+ db.setName(dbName);
+
+ dropDatabaseIfExists(dbName);
+ client.createDatabase(db);
+
+ try {
+ assertEquals(dbName, client.getDatabase(dbName).getName());
+ } finally {
+ client.dropDatabase(dbName, true, true, true);
+ }
+ }
+
+ private static void configureZooKeeperServiceDiscovery(HiveConf conf) {
+ conf.setVar(ConfVars.METASTOREURIS, zkServer.getConnectString());
+ conf.setVar(ConfVars.METASTORE_ZOOKEEPER_NAMESPACE,
+ TestRemoteHiveMetaStoreZKBindHost.class.getSimpleName());
+ conf.setVar(ConfVars.METASTORE_SERVICE_DISCOVERY_MODE, "zookeeper");
+ conf.setVar(ConfVars.METASTORE_THRIFT_BIND_HOST, "localhost");
+ }
+
+ private static void configureMetaStore(HiveConf conf) {
+ conf.setClass(ConfVars.METASTORE_EXPRESSION_PROXY_CLASS.varname,
+ MockPartitionExpressionForMetastore.class, PartitionExpressionProxy.class);
+ conf.setIntVar(ConfVars.METASTORETHRIFTCONNECTIONRETRIES, 3);
+ conf.setTimeVar(ConfVars.METASTORE_CLIENT_CONNECT_RETRY_DELAY, 100,
+ TimeUnit.MILLISECONDS);
+ conf.setTimeVar(ConfVars.METASTORE_CLIENT_CONNECTION_TIMEOUT, 5,
+ TimeUnit.SECONDS);
+ conf.setTimeVar(ConfVars.METASTORE_CLIENT_SOCKET_TIMEOUT, 30,
+ TimeUnit.SECONDS);
+ }
+
+ private static void waitForMetaStoreRegistration(HiveConf conf) throws Exception {
+ long deadline = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(30);
+ Exception lastException = null;
+ while (System.currentTimeMillis() < deadline) {
+ try {
+ List serverUris = conf.getMetastoreZKConfig().getServerUris();
+ if (!serverUris.isEmpty()) {
+ return;
+ }
+ } catch (Exception e) {
+ lastException = e;
+ }
+ Thread.sleep(250);
+ }
+
+ AssertionError error = new AssertionError("No metastore instance registered in ZooKeeper "
+ + "namespace " + conf.getVar(ConfVars.METASTORE_ZOOKEEPER_NAMESPACE));
+ if (lastException != null) {
+ error.initCause(lastException);
+ }
+ throw error;
+ }
+
+ private static void dropDatabaseIfExists(String dbName) throws Exception {
+ try {
+ client.dropDatabase(dbName, true, true, true);
+ } catch (NoSuchObjectException e) {
+ // Ignore missing databases left from previous cleanup attempts.
+ }
+ }
+}
diff --git a/pom.xml b/pom.xml
index c96fa7d27403..502e950f25ec 100644
--- a/pom.xml
+++ b/pom.xml
@@ -229,7 +229,7 @@
1.4
1.5
2.9.1
- 3.4.6
+ 3.5.8
1.1
2.4.0
2.7.1
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/zookeeper/CuratorFrameworkSingleton.java b/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/zookeeper/CuratorFrameworkSingleton.java
index b55b6cab83b5..dcfd96fe7bb8 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/zookeeper/CuratorFrameworkSingleton.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/zookeeper/CuratorFrameworkSingleton.java
@@ -18,16 +18,11 @@
package org.apache.hadoop.hive.ql.lockmgr.zookeeper;
-import java.util.concurrent.TimeUnit;
-
import org.apache.hive.common.util.ShutdownHookManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.curator.framework.CuratorFramework;
-import org.apache.curator.framework.CuratorFrameworkFactory;
-import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.hadoop.hive.conf.HiveConf;
-import org.apache.hadoop.hive.ql.util.ZooKeeperHiveHelper;
public class CuratorFrameworkSingleton {
private static HiveConf conf = null;
@@ -51,15 +46,7 @@ public static synchronized CuratorFramework getInstance(HiveConf hiveConf) {
} else {
conf = hiveConf;
}
- int sessionTimeout = (int) conf.getTimeVar(HiveConf.ConfVars.HIVE_ZOOKEEPER_SESSION_TIMEOUT, TimeUnit.MILLISECONDS);
- int baseSleepTime = (int) conf.getTimeVar(HiveConf.ConfVars.HIVE_ZOOKEEPER_CONNECTION_BASESLEEPTIME, TimeUnit.MILLISECONDS);
- int maxRetries = conf.getIntVar(HiveConf.ConfVars.HIVE_ZOOKEEPER_CONNECTION_MAX_RETRIES);
- String quorumServers = ZooKeeperHiveHelper.getQuorumServers(conf);
-
- sharedClient = CuratorFrameworkFactory.builder().connectString(quorumServers)
- .sessionTimeoutMs(sessionTimeout)
- .retryPolicy(new ExponentialBackoffRetry(baseSleepTime, maxRetries))
- .build();
+ sharedClient = conf.getZKConfig().getNewZookeeperClient();
sharedClient.start();
}
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/util/ZooKeeperHiveHelper.java b/ql/src/java/org/apache/hadoop/hive/ql/util/ZooKeeperHiveHelper.java
deleted file mode 100644
index 0e9987457d4d..000000000000
--- a/ql/src/java/org/apache/hadoop/hive/ql/util/ZooKeeperHiveHelper.java
+++ /dev/null
@@ -1,63 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.hadoop.hive.ql.util;
-
-import org.apache.hadoop.hive.conf.HiveConf;
-import org.apache.zookeeper.Watcher;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class ZooKeeperHiveHelper {
- public static final Logger LOG = LoggerFactory.getLogger(ZooKeeperHiveHelper.class.getName());
- public static final String ZOOKEEPER_PATH_SEPARATOR = "/";
-
- /**
- * Get the ensemble server addresses from the configuration. The format is: host1:port,
- * host2:port..
- *
- * @param conf
- **/
- public static String getQuorumServers(HiveConf conf) {
- String[] hosts = conf.getVar(HiveConf.ConfVars.HIVE_ZOOKEEPER_QUORUM).split(",");
- String port = conf.getVar(HiveConf.ConfVars.HIVE_ZOOKEEPER_CLIENT_PORT);
- StringBuilder quorum = new StringBuilder();
- for (int i = 0; i < hosts.length; i++) {
- quorum.append(hosts[i].trim());
- if (!hosts[i].contains(":")) {
- // if the hostname doesn't contain a port, add the configured port to hostname
- quorum.append(":");
- quorum.append(port);
- }
-
- if (i != hosts.length - 1)
- quorum.append(",");
- }
-
- return quorum.toString();
- }
-
- /**
- * A no-op watcher class
- */
- public static class DummyWatcher implements Watcher {
- @Override
- public void process(org.apache.zookeeper.WatchedEvent event) {
- }
- }
-}
diff --git a/ql/src/test/org/apache/hadoop/hive/ql/lockmgr/zookeeper/TestZookeeperLockManager.java b/ql/src/test/org/apache/hadoop/hive/ql/lockmgr/zookeeper/TestZookeeperLockManager.java
index 3f9926eea5b9..2257fd90ad97 100644
--- a/ql/src/test/org/apache/hadoop/hive/ql/lockmgr/zookeeper/TestZookeeperLockManager.java
+++ b/ql/src/test/org/apache/hadoop/hive/ql/lockmgr/zookeeper/TestZookeeperLockManager.java
@@ -18,10 +18,6 @@
package org.apache.hadoop.hive.ql.lockmgr.zookeeper;
-import java.io.File;
-import java.nio.file.Files;
-import java.nio.file.Paths;
-
import org.apache.hadoop.hive.common.metrics.MetricsTestUtils;
import org.apache.hadoop.hive.common.metrics.common.MetricsConstant;
import org.apache.hadoop.hive.common.metrics.common.MetricsFactory;
@@ -32,7 +28,6 @@
import org.apache.hadoop.hive.ql.lockmgr.HiveLockMode;
import org.apache.hadoop.hive.ql.lockmgr.HiveLockObject;
import org.apache.hadoop.hive.ql.lockmgr.HiveLockObject.HiveLockObjectData;
-import org.apache.hadoop.hive.ql.util.ZooKeeperHiveHelper;
import org.apache.zookeeper.KeeperException;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
@@ -43,9 +38,6 @@
import org.junit.After;
import org.junit.Test;
-import com.fasterxml.jackson.databind.JsonNode;
-import com.fasterxml.jackson.databind.ObjectMapper;
-
public class TestZookeeperLockManager {
private HiveConf conf;
@@ -113,15 +105,15 @@ public void testDeleteNoChildren() throws Exception
public void testGetQuorumServers() {
conf.setVar(HiveConf.ConfVars.HIVE_ZOOKEEPER_QUORUM, "node1");
conf.setVar(HiveConf.ConfVars.HIVE_ZOOKEEPER_CLIENT_PORT, "9999");
- Assert.assertEquals("node1:9999", ZooKeeperHiveHelper.getQuorumServers(conf));
+ Assert.assertEquals("node1:9999", conf.getZKConfig().getQuorumServers());
conf.setVar(HiveConf.ConfVars.HIVE_ZOOKEEPER_QUORUM, "node1,node2,node3");
conf.setVar(HiveConf.ConfVars.HIVE_ZOOKEEPER_CLIENT_PORT, "9999");
- Assert.assertEquals("node1:9999,node2:9999,node3:9999", ZooKeeperHiveHelper.getQuorumServers(conf));
+ Assert.assertEquals("node1:9999,node2:9999,node3:9999", conf.getZKConfig().getQuorumServers());
conf.setVar(HiveConf.ConfVars.HIVE_ZOOKEEPER_QUORUM, "node1:5666,node2,node3");
conf.setVar(HiveConf.ConfVars.HIVE_ZOOKEEPER_CLIENT_PORT, "9999");
- Assert.assertEquals("node1:5666,node2:9999,node3:9999", ZooKeeperHiveHelper.getQuorumServers(conf));
+ Assert.assertEquals("node1:5666,node2:9999,node3:9999", conf.getZKConfig().getQuorumServers());
}
@Test
diff --git a/service/src/java/org/apache/hive/service/server/HiveServer2.java b/service/src/java/org/apache/hive/service/server/HiveServer2.java
index e5f449122b91..854322474d44 100644
--- a/service/src/java/org/apache/hive/service/server/HiveServer2.java
+++ b/service/src/java/org/apache/hive/service/server/HiveServer2.java
@@ -19,7 +19,6 @@
package org.apache.hive.service.server;
import java.io.IOException;
-import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
@@ -39,17 +38,16 @@
import org.apache.commons.cli.ParseException;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.apache.curator.framework.CuratorFramework;
-import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.ACLProvider;
import org.apache.curator.framework.api.BackgroundCallback;
import org.apache.curator.framework.api.CuratorEvent;
import org.apache.curator.framework.api.CuratorEventType;
-import org.apache.curator.framework.recipes.nodes.PersistentEphemeralNode;
-import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.hadoop.hive.common.JvmPauseMonitor;
import org.apache.hadoop.hive.common.LogUtils;
import org.apache.hadoop.hive.common.LogUtils.LogInitializationException;
import org.apache.hadoop.hive.common.ServerUtils;
+import org.apache.hadoop.hive.common.ZKDeRegisterWatcher;
+import org.apache.hadoop.hive.common.ZooKeeperHiveHelper;
import org.apache.hadoop.hive.common.cli.CommonCliOptions;
import org.apache.hadoop.hive.common.metrics.common.MetricsFactory;
import org.apache.hadoop.hive.conf.HiveConf;
@@ -62,7 +60,6 @@
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.metadata.HiveMaterializedViewsRegistry;
import org.apache.hadoop.hive.ql.session.ClearDanglingScratchDir;
-import org.apache.hadoop.hive.ql.util.ZooKeeperHiveHelper;
import org.apache.hadoop.hive.shims.ShimLoader;
import org.apache.hadoop.hive.shims.Utils;
import org.apache.hadoop.security.UserGroupInformation;
@@ -79,8 +76,6 @@
import org.apache.hive.service.cli.thrift.ThriftHttpCLIService;
import org.apache.hive.service.servlet.QueryProfileServlet;
import org.apache.logging.log4j.util.Strings;
-import org.apache.zookeeper.CreateMode;
-import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs.Ids;
@@ -91,7 +86,6 @@
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Joiner;
-import com.google.common.base.Preconditions;
/**
* HiveServer2.
@@ -102,10 +96,7 @@ public class HiveServer2 extends CompositeService {
private static final Logger LOG = LoggerFactory.getLogger(HiveServer2.class);
private CLIService cliService;
private ThriftCLIService thriftCLIService;
- private PersistentEphemeralNode znode;
- private String znodePath;
- private CuratorFramework zooKeeperClient;
- private boolean deregisteredWithZooKeeper = false; // Set to true only when deregistration happens
+ private ZooKeeperHiveHelper zooKeeperHelper;
private HttpServer webServer; // Web UI
public HiveServer2() {
@@ -293,78 +284,22 @@ public List getAclForPath(String path) {
* @throws Exception
*/
private void addServerInstanceToZooKeeper(HiveConf hiveConf) throws Exception {
- String zooKeeperEnsemble = ZooKeeperHiveHelper.getQuorumServers(hiveConf);
- String rootNamespace = hiveConf.getVar(HiveConf.ConfVars.HIVE_SERVER2_ZOOKEEPER_NAMESPACE);
String instanceURI = getServerInstanceURI();
setUpZooKeeperAuth(hiveConf);
- int sessionTimeout =
- (int) hiveConf.getTimeVar(HiveConf.ConfVars.HIVE_ZOOKEEPER_SESSION_TIMEOUT,
- TimeUnit.MILLISECONDS);
- int baseSleepTime =
- (int) hiveConf.getTimeVar(HiveConf.ConfVars.HIVE_ZOOKEEPER_CONNECTION_BASESLEEPTIME,
- TimeUnit.MILLISECONDS);
- int maxRetries = hiveConf.getIntVar(HiveConf.ConfVars.HIVE_ZOOKEEPER_CONNECTION_MAX_RETRIES);
- // Create a CuratorFramework instance to be used as the ZooKeeper client
- // Use the zooKeeperAclProvider to create appropriate ACLs
- zooKeeperClient =
- CuratorFrameworkFactory.builder().connectString(zooKeeperEnsemble)
- .sessionTimeoutMs(sessionTimeout).aclProvider(zooKeeperAclProvider)
- .retryPolicy(new ExponentialBackoffRetry(baseSleepTime, maxRetries)).build();
- zooKeeperClient.start();
- // Create the parent znodes recursively; ignore if the parent already exists.
- try {
- zooKeeperClient.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT)
- .forPath(ZooKeeperHiveHelper.ZOOKEEPER_PATH_SEPARATOR + rootNamespace);
- LOG.info("Created the root name space: " + rootNamespace + " on ZooKeeper for HiveServer2");
- } catch (KeeperException e) {
- if (e.code() != KeeperException.Code.NODEEXISTS) {
- LOG.error("Unable to create HiveServer2 namespace: " + rootNamespace + " on ZooKeeper", e);
- throw e;
- }
- }
- // Create a znode under the rootNamespace parent for this instance of the server
- // Znode name: serverUri=host:port;version=versionInfo;sequence=sequenceNumber
- try {
- String pathPrefix =
- ZooKeeperHiveHelper.ZOOKEEPER_PATH_SEPARATOR + rootNamespace
- + ZooKeeperHiveHelper.ZOOKEEPER_PATH_SEPARATOR + "serverUri=" + instanceURI + ";"
- + "version=" + HiveVersionInfo.getVersion() + ";" + "sequence=";
- String znodeData = "";
- if (hiveConf.getBoolVar(HiveConf.ConfVars.HIVE_SERVER2_ZOOKEEPER_PUBLISH_CONFIGS)) {
- // HiveServer2 configs that this instance will publish to ZooKeeper,
- // so that the clients can read these and configure themselves properly.
- Map confsToPublish = new HashMap();
- addConfsToPublish(hiveConf, confsToPublish);
- // Publish configs for this instance as the data on the node
- znodeData = Joiner.on(';').withKeyValueSeparator("=").join(confsToPublish);
- } else {
- znodeData = instanceURI;
- }
- byte[] znodeDataUTF8 = znodeData.getBytes(Charset.forName("UTF-8"));
- znode =
- new PersistentEphemeralNode(zooKeeperClient,
- PersistentEphemeralNode.Mode.EPHEMERAL_SEQUENTIAL, pathPrefix, znodeDataUTF8);
- znode.start();
- // We'll wait for 120s for node creation
- long znodeCreationTimeout = 120;
- if (!znode.waitForInitialCreate(znodeCreationTimeout, TimeUnit.SECONDS)) {
- throw new Exception("Max znode creation wait time: " + znodeCreationTimeout + "s exhausted");
- }
- setDeregisteredWithZooKeeper(false);
- znodePath = znode.getActualPath();
- // Set a watch on the znode
- if (zooKeeperClient.checkExists().usingWatcher(new DeRegisterWatcher()).forPath(znodePath) == null) {
- // No node exists, throw exception
- throw new Exception("Unable to create znode for this HiveServer2 instance on ZooKeeper.");
- }
- LOG.info("Created a znode on ZooKeeper for HiveServer2 uri: " + instanceURI);
- } catch (Exception e) {
- LOG.error("Unable to create a znode for this server instance", e);
- if (znode != null) {
- znode.close();
- }
- throw (e);
- }
+ String znodeData;
+ if (hiveConf.getBoolVar(HiveConf.ConfVars.HIVE_SERVER2_ZOOKEEPER_PUBLISH_CONFIGS)) {
+ Map confsToPublish = new HashMap();
+ addConfsToPublish(hiveConf, confsToPublish);
+ znodeData = Joiner.on(';').withKeyValueSeparator("=").join(confsToPublish);
+ } else {
+ znodeData = instanceURI;
+ }
+ zooKeeperHelper = hiveConf.getZKConfig();
+ String znodePathPrefix = "serverUri=" + instanceURI + ";version="
+ + HiveVersionInfo.getVersion() + ";sequence=";
+ zooKeeperHelper.addServerInstanceToZooKeeper(znodePathPrefix, znodeData,
+ zooKeeperAclProvider, new DeRegisterWatcher(zooKeeperHelper));
+ LOG.info("Created a znode on ZooKeeper for HiveServer2 uri: " + instanceURI);
}
/**
@@ -429,48 +364,35 @@ private void setUpZooKeeperAuth(HiveConf hiveConf) throws Exception {
* instance is deleted. Additionally, it shuts down the server if there are no more active client
* sessions at the time of receiving a 'NodeDeleted' notification from ZooKeeper.
*/
- private class DeRegisterWatcher implements Watcher {
+
+ public class DeRegisterWatcher extends ZKDeRegisterWatcher {
+ public DeRegisterWatcher(ZooKeeperHiveHelper zooKeeperHiveHelper) {
+ super(zooKeeperHiveHelper);
+ }
+
@Override
public void process(WatchedEvent event) {
+ super.process(event);
if (event.getType().equals(Watcher.Event.EventType.NodeDeleted)) {
- if (znode != null) {
- try {
- znode.close();
- LOG.warn("This HiveServer2 instance is now de-registered from ZooKeeper. "
- + "The server will be shut down after the last client sesssion completes.");
- } catch (IOException e) {
- LOG.error("Failed to close the persistent ephemeral znode", e);
- } finally {
- HiveServer2.this.setDeregisteredWithZooKeeper(true);
- // If there are no more active client sessions, stop the server
- if (cliService.getSessionManager().getOpenSessionCount() == 0) {
- LOG.warn("This instance of HiveServer2 has been removed from the list of server "
- + "instances available for dynamic service discovery. "
- + "The last client session has ended - will shutdown now.");
- HiveServer2.this.stop();
- }
- }
+ // If there are no more active client sessions, stop the server
+ if (cliService.getSessionManager().getOpenSessionCount() == 0) {
+ LOG.warn("This instance of HiveServer2 has been removed from the list of server "
+ + "instances available for dynamic service discovery. "
+ + "The last client session has ended - will shutdown now.");
+ HiveServer2.this.stop();
}
}
}
}
private void removeServerInstanceFromZooKeeper() throws Exception {
- setDeregisteredWithZooKeeper(true);
-
- if (znode != null) {
- znode.close();
+ if (zooKeeperHelper != null) {
+ zooKeeperHelper.removeServerInstanceFromZooKeeper();
}
- zooKeeperClient.close();
- LOG.info("Server instance removed from ZooKeeper.");
}
public boolean isDeregisteredWithZooKeeper() {
- return deregisteredWithZooKeeper;
- }
-
- private void setDeregisteredWithZooKeeper(boolean deregisteredWithZooKeeper) {
- this.deregisteredWithZooKeeper = deregisteredWithZooKeeper;
+ return zooKeeperHelper != null && zooKeeperHelper.isDeregisteredWithZooKeeper();
}
private String getServerInstanceURI() throws Exception {
@@ -656,13 +578,8 @@ private static void startHiveServer2() throws Throwable {
*/
static void deleteServerInstancesFromZooKeeper(String versionNumber) throws Exception {
HiveConf hiveConf = new HiveConf();
- String zooKeeperEnsemble = ZooKeeperHiveHelper.getQuorumServers(hiveConf);
+ CuratorFramework zooKeeperClient = hiveConf.getZKConfig().getNewZookeeperClient();
String rootNamespace = hiveConf.getVar(HiveConf.ConfVars.HIVE_SERVER2_ZOOKEEPER_NAMESPACE);
- int baseSleepTime = (int) hiveConf.getTimeVar(HiveConf.ConfVars.HIVE_ZOOKEEPER_CONNECTION_BASESLEEPTIME, TimeUnit.MILLISECONDS);
- int maxRetries = hiveConf.getIntVar(HiveConf.ConfVars.HIVE_ZOOKEEPER_CONNECTION_MAX_RETRIES);
- CuratorFramework zooKeeperClient =
- CuratorFrameworkFactory.builder().connectString(zooKeeperEnsemble)
- .retryPolicy(new ExponentialBackoffRetry(baseSleepTime, maxRetries)).build();
zooKeeperClient.start();
List znodePaths =
zooKeeperClient.getChildren().forPath(
diff --git a/shims/common/src/main/java/org/apache/hadoop/hive/thrift/HiveDelegationTokenManager.java b/shims/common/src/main/java/org/apache/hadoop/hive/thrift/HiveDelegationTokenManager.java
index 7b0c2ed771b0..21ae0de0bfba 100644
--- a/shims/common/src/main/java/org/apache/hadoop/hive/thrift/HiveDelegationTokenManager.java
+++ b/shims/common/src/main/java/org/apache/hadoop/hive/thrift/HiveDelegationTokenManager.java
@@ -67,6 +67,16 @@ public class HiveDelegationTokenManager {
"hive.cluster.delegation.token.store.zookeeper.acl";
public static final String DELEGATION_TOKEN_STORE_ZK_ZNODE_DEFAULT =
"/hivedelegation";
+ public static final String DELEGATION_TOKEN_STORE_ZK_SSL_ENABLE =
+ "hive.cluster.delegation.token.store.zookeeper.ssl.client.enable";
+ public static final String DELEGATION_TOKEN_STORE_ZK_KEYSTORE_LOCATION =
+ "hive.cluster.delegation.token.store.zookeeper.keystore.location";
+ public static final String DELEGATION_TOKEN_STORE_ZK_KEYSTORE_PASSWORD =
+ "hive.cluster.delegation.token.store.zookeeper.keystore.password";
+ public static final String DELEGATION_TOKEN_STORE_ZK_TRUSTSTORE_LOCATION =
+ "hive.cluster.delegation.token.store.zookeeper.truststore.location";
+ public static final String DELEGATION_TOKEN_STORE_ZK_TRUSTSTORE_PASSWORD =
+ "hive.cluster.delegation.token.store.zookeeper.truststore.password";
protected DelegationTokenSecretManager secretManager;
diff --git a/shims/common/src/main/java/org/apache/hadoop/hive/thrift/ZooKeeperTokenStore.java b/shims/common/src/main/java/org/apache/hadoop/hive/thrift/ZooKeeperTokenStore.java
index 463310b77d3a..31df683db787 100644
--- a/shims/common/src/main/java/org/apache/hadoop/hive/thrift/ZooKeeperTokenStore.java
+++ b/shims/common/src/main/java/org/apache/hadoop/hive/thrift/ZooKeeperTokenStore.java
@@ -24,6 +24,7 @@
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.TimeUnit;
import org.apache.commons.lang3.StringUtils;
import org.apache.curator.framework.CuratorFramework;
@@ -31,6 +32,7 @@
import org.apache.curator.framework.api.ACLProvider;
import org.apache.curator.framework.imps.CuratorFrameworkState;
import org.apache.curator.retry.ExponentialBackoffRetry;
+import org.apache.curator.utils.ZookeeperFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.shims.ShimLoader;
import org.apache.hadoop.hive.shims.Utils;
@@ -40,9 +42,13 @@
import org.apache.hadoop.security.token.delegation.HiveDelegationTokenSupport;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooDefs.Perms;
+import org.apache.zookeeper.client.ZKClientConfig;
+import org.apache.zookeeper.common.ClientX509Util;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Id;
import org.slf4j.Logger;
@@ -64,6 +70,11 @@ public class ZooKeeperTokenStore implements DelegationTokenStore {
private volatile CuratorFramework zkSession;
private String zkConnectString;
private int connectTimeoutMillis;
+ private boolean sslEnabled;
+ private String keyStoreLocation;
+ private String keyStorePassword;
+ private String trustStoreLocation;
+ private String trustStorePassword;
private List newNodeAcl = Arrays.asList(new ACL(Perms.ALL, Ids.AUTH_IDS));
/**
@@ -90,6 +101,21 @@ public List getAclForPath(String path) {
private Configuration conf;
+ private static final String HIVE_ZOOKEEPER_CLIENT_PORT =
+ "hive.zookeeper.client.port";
+ private static final String HIVE_ZOOKEEPER_CONNECTION_TIMEOUT =
+ "hive.zookeeper.connection.timeout";
+ private static final String HIVE_ZOOKEEPER_SSL_ENABLE =
+ "hive.zookeeper.ssl.client.enable";
+ private static final String HIVE_ZOOKEEPER_SSL_KEYSTORE_LOCATION =
+ "hive.zookeeper.ssl.keystore.location";
+ private static final String HIVE_ZOOKEEPER_SSL_KEYSTORE_PASSWORD =
+ "hive.zookeeper.ssl.keystore.password";
+ private static final String HIVE_ZOOKEEPER_SSL_TRUSTSTORE_LOCATION =
+ "hive.zookeeper.ssl.truststore.location";
+ private static final String HIVE_ZOOKEEPER_SSL_TRUSTSTORE_PASSWORD =
+ "hive.zookeeper.ssl.truststore.password";
+
/**
* Default constructor for dynamic instantiation w/ Configurable
* (ReflectionUtils does not support Configuration constructor injection).
@@ -97,14 +123,19 @@ public List getAclForPath(String path) {
protected ZooKeeperTokenStore() {
}
- private CuratorFramework getSession() {
+ public CuratorFramework getSession() {
if (zkSession == null || zkSession.getState() == CuratorFrameworkState.STOPPED) {
synchronized (this) {
if (zkSession == null || zkSession.getState() == CuratorFrameworkState.STOPPED) {
- zkSession =
- CuratorFrameworkFactory.builder().connectString(zkConnectString)
- .connectionTimeoutMs(connectTimeoutMillis).aclProvider(aclDefaultProvider)
- .retryPolicy(new ExponentialBackoffRetry(1000, 3)).build();
+ CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder()
+ .connectString(zkConnectString)
+ .connectionTimeoutMs(connectTimeoutMillis)
+ .aclProvider(aclDefaultProvider)
+ .retryPolicy(new ExponentialBackoffRetry(1000, 3));
+ if (sslEnabled) {
+ builder = builder.zookeeperFactory(new TokenStoreSSLZookeeperFactory());
+ }
+ zkSession = builder.build();
zkSession.start();
}
}
@@ -450,11 +481,35 @@ public void init(Object hmsHandler, ServerMode smode) {
+ HiveDelegationTokenManager.DELEGATION_TOKEN_STORE_ZK_CONNECT_STR_ALTERNATE
+ WHEN_ZK_DSTORE_MSG);
}
+ String zkConnectPort = conf.get(HIVE_ZOOKEEPER_CLIENT_PORT, "2181");
+ zkConnectString = appendPortToConnectString(zkConnectString, zkConnectPort);
+ connectTimeoutMillis = (int) conf.getTimeDuration(
+ HIVE_ZOOKEEPER_CONNECTION_TIMEOUT, 15000, TimeUnit.MILLISECONDS);
+ sslEnabled = conf.getBoolean(HIVE_ZOOKEEPER_SSL_ENABLE, false);
+ if (sslEnabled) {
+ keyStoreLocation = conf.get(HIVE_ZOOKEEPER_SSL_KEYSTORE_LOCATION, "");
+ keyStorePassword = getPasswordString(HIVE_ZOOKEEPER_SSL_KEYSTORE_PASSWORD);
+ trustStoreLocation = conf.get(HIVE_ZOOKEEPER_SSL_TRUSTSTORE_LOCATION, "");
+ trustStorePassword = getPasswordString(HIVE_ZOOKEEPER_SSL_TRUSTSTORE_PASSWORD);
+ }
+ } else {
+ connectTimeoutMillis =
+ conf.getInt(
+ HiveDelegationTokenManager.DELEGATION_TOKEN_STORE_ZK_CONNECT_TIMEOUTMILLIS,
+ CuratorFrameworkFactory.builder().getConnectionTimeoutMs());
+ sslEnabled = conf.getBoolean(
+ HiveDelegationTokenManager.DELEGATION_TOKEN_STORE_ZK_SSL_ENABLE, false);
+ if (sslEnabled) {
+ keyStoreLocation = conf.get(
+ HiveDelegationTokenManager.DELEGATION_TOKEN_STORE_ZK_KEYSTORE_LOCATION, "");
+ keyStorePassword = getPasswordString(
+ HiveDelegationTokenManager.DELEGATION_TOKEN_STORE_ZK_KEYSTORE_PASSWORD);
+ trustStoreLocation = conf.get(
+ HiveDelegationTokenManager.DELEGATION_TOKEN_STORE_ZK_TRUSTSTORE_LOCATION, "");
+ trustStorePassword = getPasswordString(
+ HiveDelegationTokenManager.DELEGATION_TOKEN_STORE_ZK_TRUSTSTORE_PASSWORD);
+ }
}
- connectTimeoutMillis =
- conf.getInt(
- HiveDelegationTokenManager.DELEGATION_TOKEN_STORE_ZK_CONNECT_TIMEOUTMILLIS,
- CuratorFrameworkFactory.builder().getConnectionTimeoutMs());
String aclStr = conf.get(HiveDelegationTokenManager.DELEGATION_TOKEN_STORE_ZK_ACL, null);
if (StringUtils.isNotBlank(aclStr)) {
this.newNodeAcl = parseACLs(aclStr);
@@ -473,4 +528,52 @@ public void init(Object hmsHandler, ServerMode smode) {
initClientAndPaths();
}
+ private String getPasswordString(String key) {
+ try {
+ char[] password = conf.getPassword(key);
+ return password == null ? null : new String(password);
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to read zookeeper configuration passwords", e);
+ }
+ }
+
+ private String appendPortToConnectString(String connectString, String clientPort) {
+ String[] hosts = connectString.split(",");
+ StringBuilder quorumServers = new StringBuilder();
+ for (int i = 0; i < hosts.length; i++) {
+ quorumServers.append(hosts[i].trim());
+ if (!hosts[i].contains(":")) {
+ quorumServers.append(":");
+ quorumServers.append(clientPort);
+ }
+ if (i != hosts.length - 1) {
+ quorumServers.append(",");
+ }
+ }
+ return quorumServers.toString();
+ }
+
+ private class TokenStoreSSLZookeeperFactory implements ZookeeperFactory {
+ @Override
+ public ZooKeeper newZooKeeper(String connectString, int sessionTimeout, Watcher watcher,
+ boolean canBeReadOnly) throws Exception {
+ ZKClientConfig clientConfig = new ZKClientConfig();
+ clientConfig.setProperty(ZKClientConfig.SECURE_CLIENT, "true");
+ clientConfig.setProperty(ZKClientConfig.ZOOKEEPER_CLIENT_CNXN_SOCKET,
+ "org.apache.zookeeper.ClientCnxnSocketNetty");
+
+ ClientX509Util x509Util = new ClientX509Util();
+ clientConfig.setProperty(x509Util.getSslKeystoreLocationProperty(),
+ StringUtils.defaultString(keyStoreLocation, ""));
+ clientConfig.setProperty(x509Util.getSslKeystorePasswdProperty(),
+ StringUtils.defaultString(keyStorePassword, ""));
+ clientConfig.setProperty(x509Util.getSslTruststoreLocationProperty(),
+ StringUtils.defaultString(trustStoreLocation, ""));
+ clientConfig.setProperty(x509Util.getSslTruststorePasswdProperty(),
+ StringUtils.defaultString(trustStorePassword, ""));
+
+ return new ZooKeeper(connectString, sessionTimeout, watcher, canBeReadOnly, clientConfig);
+ }
+ }
+
}