You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by se...@apache.org on 2017/09/01 18:48:03 UTC
[1/2] hive git commit: HIVE-17409 : refactor LLAP ZK registry to make
the ZK-registry part reusable (Sergey Shelukhin,
reviewed by Prasanth Jayachandran)
Repository: hive
Updated Branches:
refs/heads/master 2e226d22f -> 4795b9969
http://git-wip-us.apache.org/repos/asf/hive/blob/4795b996/llap-client/src/java/org/apache/hadoop/hive/registry/impl/ZkRegistryBase.java
----------------------------------------------------------------------
diff --git a/llap-client/src/java/org/apache/hadoop/hive/registry/impl/ZkRegistryBase.java b/llap-client/src/java/org/apache/hadoop/hive/registry/impl/ZkRegistryBase.java
new file mode 100644
index 0000000..c773770
--- /dev/null
+++ b/llap-client/src/java/org/apache/hadoop/hive/registry/impl/ZkRegistryBase.java
@@ -0,0 +1,549 @@
+/*
+ * Licensed 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.registry.impl;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+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;
+import org.apache.curator.framework.recipes.cache.PathChildrenCache;
+import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
+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;
+import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
+import org.apache.hadoop.hive.llap.LlapUtil;
+import org.apache.hadoop.hive.registry.ServiceInstance;
+import org.apache.hadoop.hive.registry.ServiceInstanceStateChangeListener;
+import org.apache.hadoop.registry.client.binding.RegistryUtils;
+import org.apache.hadoop.registry.client.binding.RegistryUtils.ServiceRecordMarshal;
+import org.apache.hadoop.registry.client.types.ServiceRecord;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.zookeeper.KeeperException.InvalidACLException;
+import org.apache.zookeeper.ZooDefs;
+import org.apache.zookeeper.data.ACL;
+import org.apache.zookeeper.data.Id;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This is currently used for implementation inheritance only; it doesn't provide a unified flow
+ * into which one can just plug a few abstract method implementations, because providing one with
+ * getInstance method is a huge pain involving lots of generics. Also, different registries may
+ * have slightly different usage patterns anyway and noone would use a registry without knowing
+ * what type it is. So, it's mostly a grab bag of methods used by ServiceInstanceSet and other
+ * parts of each implementation.
+ */
+public abstract class ZkRegistryBase<InstanceType extends ServiceInstance> {
+ private static final Logger LOG = LoggerFactory.getLogger(ZkRegistryBase.class);
+ private final static String SASL_NAMESPACE = "sasl";
+ private final static String UNSECURE_NAMESPACE = "unsecure";
+
+ static final String UNIQUE_IDENTIFIER = "registry.unique.id";
+ private static final UUID uniq = UUID.randomUUID();
+
+ protected final Configuration conf;
+ protected final CuratorFramework zooKeeperClient;
+ // userPathPrefix is the path specific to the user for which ACLs should be restrictive.
+ // workersPath is the directory path where all the worker znodes are located.
+ protected final String workersPath;
+ private final String userPathPrefix, workerNodePrefix;
+
+ protected final ServiceRecordMarshal encoder; // to marshal/unmarshal znode data
+
+ private final Set<ServiceInstanceStateChangeListener<InstanceType>> stateChangeListeners;
+
+ private final boolean doCheckAcls;
+ // Secure ZK is only set up by the registering service; anyone can read the registrations.
+ private final String zkPrincipal, zkKeytab, saslLoginContextName;
+ private String userNameFromPrincipal; // Only set when setting up the secure config for ZK.
+ private final String disableMessage;
+
+ private final Lock instanceCacheLock = new ReentrantLock();
+ private final Map<String, Set<InstanceType>> pathToInstanceCache;
+ private final Map<String, Set<InstanceType>> nodeToInstanceCache;
+
+ // The registration znode.
+ private PersistentEphemeralNode znode;
+ private String znodePath; // unique identity for this instance
+
+ private PathChildrenCache instancesCache; // Created on demand.
+
+ /** Local hostname. */
+ protected static final String hostname;
+ static {
+ String localhost = "localhost";
+ try {
+ localhost = InetAddress.getLocalHost().getCanonicalHostName();
+ } catch (UnknownHostException uhe) {
+ // ignore
+ }
+ hostname = localhost;
+ }
+
+ /**
+ * @param rootNs A single root namespace override. Not recommended.
+ * @param nsPrefix The namespace prefix to use with default namespaces.
+ * @param userScopePathPrefix The prefix to use for the user-specific part of the path.
+ * @param workerPrefix The prefix to use for each worker znode.
+ * @param zkSaslLoginContextName SASL login context name for ZK security; null if not needed.
+ * @param zkPrincipal ZK security principal.
+ * @param zkKeytab ZK security keytab.
+ * @param aclsConfig A config setting to use to determine if ACLs should be verified.
+ */
+ public ZkRegistryBase(String instanceName, Configuration conf, String rootNs, String nsPrefix,
+ String userScopePathPrefix, String workerPrefix,
+ String zkSaslLoginContextName, String zkPrincipal, String zkKeytab, ConfVars aclsConfig) {
+ this.conf = new Configuration(conf);
+ this.saslLoginContextName = zkSaslLoginContextName;
+ this.zkPrincipal = zkPrincipal;
+ this.zkKeytab = zkKeytab;
+ if (aclsConfig != null) {
+ this.doCheckAcls = HiveConf.getBoolVar(conf, aclsConfig);
+ this.disableMessage = "Set " + aclsConfig.varname + " to false to disable ACL validation";
+ } else {
+ this.doCheckAcls = true;
+ this.disableMessage = "";
+ }
+ 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
+ // worker does not respond due to communication interruptions it will retain the same sequence
+ // number when it returns back. If session timeout expires, the node will be deleted and new
+ // addition of the same node (restart) will get next sequence number
+ this.userPathPrefix = userScopePathPrefix + getZkPathUser(this.conf);
+ this.workerNodePrefix = workerPrefix;
+ this.workersPath = "/" + userPathPrefix + "/" + instanceName + "/workers";
+ this.instancesCache = null;
+ this.stateChangeListeners = new HashSet<>();
+ this.pathToInstanceCache = new ConcurrentHashMap<>();
+ this.nodeToInstanceCache = new ConcurrentHashMap<>();
+
+ final boolean isSecure = UserGroupInformation.isSecurityEnabled();
+ ACLProvider zooKeeperAclProvider = new ACLProvider() {
+ @Override
+ public List<ACL> getDefaultAcl() {
+ // We always return something from getAclForPath so this should not happen.
+ LOG.warn("getDefaultAcl was called");
+ return Lists.newArrayList(ZooDefs.Ids.OPEN_ACL_UNSAFE);
+ }
+
+ @Override
+ public List<ACL> getAclForPath(String path) {
+ if (!isSecure || path == null || !path.contains(userPathPrefix)) {
+ // No security or the path is below the user path - full access.
+ return Lists.newArrayList(ZooDefs.Ids.OPEN_ACL_UNSAFE);
+ }
+ return createSecureAcls();
+ }
+ };
+ if (rootNs == null) {
+ rootNs = nsPrefix + (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();
+ }
+
+ private static List<ACL> createSecureAcls() {
+ // Read all to the world
+ List<ACL> nodeAcls = new ArrayList<ACL>(ZooDefs.Ids.READ_ACL_UNSAFE);
+ // Create/Delete/Write/Admin to creator
+ nodeAcls.addAll(ZooDefs.Ids.CREATOR_ALL_ACL);
+ return nodeAcls;
+ }
+
+ /**
+ * Get the ensemble server addresses from the configuration. The format is: host1:port,
+ * host2:port..
+ *
+ * @param conf
+ **/
+ private String getQuorumServers(Configuration conf) {
+ String[] hosts = conf.getTrimmedStrings(ConfVars.HIVE_ZOOKEEPER_QUORUM.varname);
+ String port = conf.get(ConfVars.HIVE_ZOOKEEPER_CLIENT_PORT.varname,
+ ConfVars.HIVE_ZOOKEEPER_CLIENT_PORT.getDefaultValue());
+ 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();
+ }
+
+ protected abstract String getZkPathUser(Configuration conf);
+
+ protected final String registerServiceRecord(ServiceRecord srv) throws IOException {
+ // restart sensitive instance id
+ srv.set(UNIQUE_IDENTIFIER, uniq.toString());
+
+ // Create a znode under the rootNamespace parent for this instance of the server
+ try {
+ // PersistentEphemeralNode will make sure the ephemeral node created on server will be present
+ // even under connection or session interruption (will automatically handle retries)
+ znode = new PersistentEphemeralNode(zooKeeperClient, Mode.EPHEMERAL_SEQUENTIAL,
+ workersPath + "/" + workerNodePrefix, encoder.toBytes(srv));
+
+ // start the creation of znodes
+ 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");
+ }
+
+ znodePath = znode.getActualPath();
+
+ if (doCheckAcls) {
+ try {
+ checkAndSetAcls();
+ } catch (Exception ex) {
+ throw new IOException("Error validating or setting ACLs. " + disableMessage, ex);
+ }
+ }
+ if (zooKeeperClient.checkExists().forPath(znodePath) == null) {
+ // No node exists, throw exception
+ throw new Exception("Unable to create znode for this instance on ZooKeeper.");
+ }
+ } catch (Exception e) {
+ LOG.error("Unable to create a znode for this server instance", e);
+ CloseableUtils.closeQuietly(znode);
+ throw (e instanceof IOException) ? (IOException)e : new IOException(e);
+ }
+ return uniq.toString();
+ }
+
+
+ private void checkAndSetAcls() throws Exception {
+ if (!UserGroupInformation.isSecurityEnabled()) return;
+ // We are trying to check ACLs on the "workers" directory, which noone except us should be
+ // able to write to. Higher-level directories shouldn't matter - we don't read them.
+ String pathToCheck = workersPath;
+ List<ACL> acls = zooKeeperClient.getACL().forPath(pathToCheck);
+ if (acls == null || acls.isEmpty()) {
+ // Can there be no ACLs? There's some access (to get ACLs), so assume it means free for all.
+ LOG.warn("No ACLs on " + pathToCheck + "; setting up ACLs. " + disableMessage);
+ setUpAcls(pathToCheck);
+ return;
+ }
+ // This could be brittle.
+ assert userNameFromPrincipal != null;
+ Id currentUser = new Id("sasl", userNameFromPrincipal);
+ for (ACL acl : acls) {
+ if ((acl.getPerms() & ~ZooDefs.Perms.READ) == 0 || currentUser.equals(acl.getId())) {
+ continue; // Read permission/no permissions, or the expected user.
+ }
+ LOG.warn("The ACL " + acl + " is unnacceptable for " + pathToCheck
+ + "; setting up ACLs. " + disableMessage);
+ setUpAcls(pathToCheck);
+ return;
+ }
+ }
+
+ private void setUpAcls(String path) throws Exception {
+ List<ACL> acls = createSecureAcls();
+ LinkedList<String> paths = new LinkedList<>();
+ paths.add(path);
+ while (!paths.isEmpty()) {
+ String currentPath = paths.poll();
+ List<String> children = zooKeeperClient.getChildren().forPath(currentPath);
+ if (children != null) {
+ for (String child : children) {
+ paths.add(currentPath + "/" + child);
+ }
+ }
+ zooKeeperClient.setACL().withACL(acls).forPath(currentPath);
+ }
+ }
+
+ private void addToCache(String path, String host, InstanceType instance) {
+ instanceCacheLock.lock();
+ try {
+ putInCache(path, pathToInstanceCache, instance);
+ putInCache(host, nodeToInstanceCache, instance);
+ } finally {
+ instanceCacheLock.unlock();
+ }
+ LOG.debug("Added path={}, host={} instance={} to cache."
+ + " pathToInstanceCache:size={}, nodeToInstanceCache:size={}",
+ path, host, instance, pathToInstanceCache.size(), nodeToInstanceCache.size());
+ }
+
+ private void removeFromCache(String path, String host) {
+ instanceCacheLock.lock();
+ try {
+ pathToInstanceCache.remove(path);
+ nodeToInstanceCache.remove(host);
+ } finally {
+ instanceCacheLock.unlock();
+ }
+ LOG.debug("Removed path={}, host={} from cache."
+ + " pathToInstanceCache:size={}, nodeToInstanceCache:size={}",
+ path, host, pathToInstanceCache.size(), nodeToInstanceCache.size());
+ }
+
+ private void putInCache(String key, Map<String, Set<InstanceType>> cache,
+ InstanceType instance) {
+ Set<InstanceType> instanceSet = cache.get(key);
+ if (instanceSet == null) {
+ instanceSet = Sets.newHashSet();
+ cache.put(key, instanceSet);
+ }
+ instanceSet.add(instance);
+ }
+
+ protected final void populateCache(PathChildrenCache instancesCache) {
+ for (ChildData childData : instancesCache.getCurrentData()) {
+ byte[] data = getWorkerData(childData, workerNodePrefix);
+ if (data == null) continue;
+ try {
+ ServiceRecord srv = encoder.fromBytes(childData.getPath(), data);
+ InstanceType instance = createServiceInstance(srv);
+ addToCache(childData.getPath(), instance.getHost(), instance);
+ } catch (IOException e) {
+ LOG.error("Unable to decode data for zkpath: {}." +
+ " Ignoring from current instances list..", childData.getPath());
+ }
+ }
+ }
+
+ protected abstract InstanceType createServiceInstance(ServiceRecord srv) throws IOException;
+
+ protected static final byte[] getWorkerData(ChildData childData, String workerNodePrefix) {
+ if (childData == null) return null;
+ byte[] data = childData.getData();
+ if (data == null) return null;
+ if (!extractNodeName(childData).startsWith(workerNodePrefix)) return null;
+ return data;
+ }
+
+ private class InstanceStateChangeListener implements PathChildrenCacheListener {
+ private final Logger LOG = LoggerFactory.getLogger(InstanceStateChangeListener.class);
+
+ @Override
+ public void childEvent(final CuratorFramework client, final PathChildrenCacheEvent event)
+ throws Exception {
+ Preconditions.checkArgument(client != null
+ && client.getState() == CuratorFrameworkState.STARTED, "client is not started");
+
+ synchronized (this) {
+ ChildData childData = event.getData();
+ if (childData == null) return;
+ String nodeName = extractNodeName(childData);
+ if (!nodeName.startsWith(workerNodePrefix)) return;
+ LOG.info("{} for zknode {}", event.getType(), childData.getPath());
+ InstanceType instance = extractServiceInstance(event, childData);
+ switch (event.getType()) {
+ case CHILD_ADDED:
+ addToCache(childData.getPath(), instance.getHost(), instance);
+ for (ServiceInstanceStateChangeListener<InstanceType> listener : stateChangeListeners) {
+ listener.onCreate(instance);
+ }
+ break;
+ case CHILD_UPDATED:
+ addToCache(childData.getPath(), instance.getHost(), instance);
+ for (ServiceInstanceStateChangeListener<InstanceType> listener : stateChangeListeners) {
+ listener.onUpdate(instance);
+ }
+ break;
+ case CHILD_REMOVED:
+ removeFromCache(childData.getPath(), instance.getHost());
+ for (ServiceInstanceStateChangeListener<InstanceType> listener : stateChangeListeners) {
+ listener.onRemove(instance);
+ }
+ break;
+ default:
+ // Ignore all the other events; logged above.
+ }
+ }
+ }
+ }
+
+ // The real implementation for the instanceset... instanceset has its own copy of the
+ // ZK cache yet completely depends on the parent in every other aspect and is thus unneeded.
+
+ public int size() {
+ // not using the path child cache here as there could be more than 1 path per host (worker and slot znodes)
+ return nodeToInstanceCache.size();
+ }
+
+ protected final Set<InstanceType> getByHost(String host) {
+ Set<InstanceType> byHost = nodeToInstanceCache.get(host);
+ byHost = (byHost == null) ? Sets.<InstanceType>newHashSet() : byHost;
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Returning " + byHost.size() + " hosts for locality allocation on " + host);
+ }
+ return byHost;
+ }
+
+ protected final Collection<InstanceType> getAll() {
+ Set<InstanceType> instances = new HashSet<>();
+ for(Set<InstanceType> instanceSet : pathToInstanceCache.values()) {
+ instances.addAll(instanceSet);
+ }
+ return instances;
+ }
+
+ private static String extractNodeName(ChildData childData) {
+ String nodeName = childData.getPath();
+ int ix = nodeName.lastIndexOf("/");
+ if (ix >= 0) {
+ nodeName = nodeName.substring(ix + 1);
+ }
+ return nodeName;
+ }
+
+ private InstanceType extractServiceInstance(
+ PathChildrenCacheEvent event, ChildData childData) {
+ byte[] data = childData.getData();
+ if (data == null) return null;
+ try {
+ ServiceRecord srv = encoder.fromBytes(event.getData().getPath(), data);
+ return createServiceInstance(srv);
+ } catch (IOException e) {
+ LOG.error("Unable to decode data for zknode: {}." +
+ " Dropping notification of type: {}", childData.getPath(), event.getType());
+ return null;
+ }
+ }
+
+ public synchronized void registerStateChangeListener(
+ ServiceInstanceStateChangeListener<InstanceType> listener) throws IOException {
+ ensureInstancesCache(0);
+ this.stateChangeListeners.add(listener);
+ }
+
+ @SuppressWarnings("resource") // Bogus warnings despite closeQuietly.
+ protected final synchronized PathChildrenCache ensureInstancesCache(
+ long clusterReadyTimeoutMs) throws IOException {
+ Preconditions.checkArgument(zooKeeperClient != null &&
+ zooKeeperClient.getState() == CuratorFrameworkState.STARTED, "client is not started");
+ // lazily create PathChildrenCache
+ PathChildrenCache instancesCache = this.instancesCache;
+ if (instancesCache != null) return instancesCache;
+ ExecutorService tp = Executors.newFixedThreadPool(1, new ThreadFactoryBuilder()
+ .setDaemon(true).setNameFormat("StateChangeNotificationHandler").build());
+ long startTimeNs = System.nanoTime(), deltaNs = clusterReadyTimeoutMs * 1000000L;
+ long sleepTimeMs = Math.min(16, clusterReadyTimeoutMs);
+ while (true) {
+ instancesCache = new PathChildrenCache(zooKeeperClient, workersPath, true);
+ instancesCache.getListenable().addListener(new InstanceStateChangeListener(), tp);
+ try {
+ instancesCache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);
+ this.instancesCache = instancesCache;
+ return instancesCache;
+ } catch (InvalidACLException e) {
+ // PathChildrenCache tried to mkdir when the znode wasn't there, and failed.
+ CloseableUtils.closeQuietly(instancesCache);
+ long elapsedNs = System.nanoTime() - startTimeNs;
+ if (deltaNs == 0 || deltaNs <= elapsedNs) {
+ LOG.error("Unable to start curator PathChildrenCache", e);
+ throw new IOException(e);
+ }
+ LOG.warn("The cluster is not started yet (InvalidACL); will retry");
+ try {
+ Thread.sleep(Math.min(sleepTimeMs, (deltaNs - elapsedNs)/1000000L));
+ } catch (InterruptedException e1) {
+ LOG.error("Interrupted while retrying the PathChildrenCache startup");
+ throw new IOException(e1);
+ }
+ sleepTimeMs = sleepTimeMs << 1;
+ } catch (Exception e) {
+ CloseableUtils.closeQuietly(instancesCache);
+ LOG.error("Unable to start curator PathChildrenCache", e);
+ throw new IOException(e);
+ }
+ }
+ }
+
+ public void start() throws IOException {
+ if (zooKeeperClient != null) {
+ String principal = ZookeeperUtils.setupZookeeperAuth(
+ conf, saslLoginContextName, zkPrincipal, zkKeytab);
+ if (principal != null) {
+ userNameFromPrincipal = LlapUtil.getUserNameFromPrincipal(principal);
+ }
+ zooKeeperClient.start();
+ }
+ // Init closeable utils in case register is not called (see HIVE-13322)
+ CloseableUtils.class.getName();
+ }
+
+ public void stop() throws IOException {
+ CloseableUtils.closeQuietly(znode);
+ CloseableUtils.closeQuietly(instancesCache);
+ CloseableUtils.closeQuietly(zooKeeperClient);
+ }
+
+ protected final Set<InstanceType> getInstancesByPath(String path) {
+ return pathToInstanceCache.get(path);
+ }
+
+ protected final String getRegistrationZnodePath() {
+ return znodePath;
+ }
+}
http://git-wip-us.apache.org/repos/asf/hive/blob/4795b996/llap-client/src/java/org/apache/hadoop/hive/registry/impl/ZookeeperUtils.java
----------------------------------------------------------------------
diff --git a/llap-client/src/java/org/apache/hadoop/hive/registry/impl/ZookeeperUtils.java b/llap-client/src/java/org/apache/hadoop/hive/registry/impl/ZookeeperUtils.java
new file mode 100644
index 0000000..454d503
--- /dev/null
+++ b/llap-client/src/java/org/apache/hadoop/hive/registry/impl/ZookeeperUtils.java
@@ -0,0 +1,116 @@
+/*
+ * Licensed 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.registry.impl;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+import javax.security.auth.login.AppConfigurationEntry;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.security.SecurityUtil;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.security.authentication.util.KerberosUtil;
+import org.apache.zookeeper.client.ZooKeeperSaslClient;
+
+public class ZookeeperUtils {
+ private static final Logger LOG = LoggerFactory.getLogger(ZookeeperUtils.class);
+
+ public static String setupZookeeperAuth(Configuration conf, String saslLoginContextName,
+ String zkPrincipal, String zkKeytab) throws IOException {
+ // If the login context name is not set, we are in the client and don't need auth.
+ if (UserGroupInformation.isSecurityEnabled() && saslLoginContextName != null) {
+ LOG.info("UGI security is enabled. Setting up ZK auth.");
+
+ if (zkPrincipal == null || zkPrincipal.isEmpty()) {
+ throw new IOException("Kerberos principal is empty");
+ }
+
+ if (zkKeytab == null || zkKeytab.isEmpty()) {
+ throw new IOException("Kerberos keytab is empty");
+ }
+
+ // Install the JAAS Configuration for the runtime
+ return setZookeeperClientKerberosJaasConfig(saslLoginContextName, zkPrincipal, zkKeytab);
+ } else {
+ LOG.info("UGI security is not enabled, or no SASL context name. " +
+ "Skipping setting up ZK auth.");
+ return null;
+ }
+ }
+
+ /**
+ * Dynamically sets up the JAAS configuration that uses kerberos
+ *
+ * @param principal
+ * @param keyTabFile
+ * @throws IOException
+ */
+ private static String setZookeeperClientKerberosJaasConfig(
+ String saslLoginContextName, String zkPrincipal, String zkKeytab) throws IOException {
+ // ZooKeeper property name to pick the correct JAAS conf section
+ System.setProperty(ZooKeeperSaslClient.LOGIN_CONTEXT_NAME_KEY, saslLoginContextName);
+
+ String principal = SecurityUtil.getServerPrincipal(zkPrincipal, "0.0.0.0");
+ JaasConfiguration jaasConf = new JaasConfiguration(
+ saslLoginContextName, principal, zkKeytab);
+
+ // Install the Configuration in the runtime.
+ javax.security.auth.login.Configuration.setConfiguration(jaasConf);
+ return principal;
+ }
+
+ /**
+ * A JAAS configuration for ZooKeeper clients intended to use for SASL
+ * Kerberos.
+ */
+ private static class JaasConfiguration extends javax.security.auth.login.Configuration {
+ // Current installed Configuration
+ private final javax.security.auth.login.Configuration baseConfig = javax.security.auth.login.Configuration
+ .getConfiguration();
+ private final String loginContextName;
+ private final String principal;
+ private final String keyTabFile;
+
+ public JaasConfiguration(String loginContextName, String principal, String keyTabFile) {
+ this.loginContextName = loginContextName;
+ this.principal = principal;
+ this.keyTabFile = keyTabFile;
+ }
+
+ @Override
+ public AppConfigurationEntry[] getAppConfigurationEntry(String appName) {
+ if (loginContextName.equals(appName)) {
+ Map<String, String> krbOptions = new HashMap<String, String>();
+ krbOptions.put("doNotPrompt", "true");
+ krbOptions.put("storeKey", "true");
+ krbOptions.put("useKeyTab", "true");
+ krbOptions.put("principal", principal);
+ krbOptions.put("keyTab", keyTabFile);
+ krbOptions.put("refreshKrb5Config", "true");
+ AppConfigurationEntry zooKeeperClientEntry = new AppConfigurationEntry(
+ KerberosUtil.getKrb5LoginModuleName(),
+ AppConfigurationEntry.LoginModuleControlFlag.REQUIRED, krbOptions);
+ return new AppConfigurationEntry[] { zooKeeperClientEntry };
+ }
+ // Try the base config
+ if (baseConfig != null) {
+ return baseConfig.getAppConfigurationEntry(appName);
+ }
+ return null;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/hive/blob/4795b996/llap-ext-client/src/java/org/apache/hadoop/hive/llap/LlapBaseInputFormat.java
----------------------------------------------------------------------
diff --git a/llap-ext-client/src/java/org/apache/hadoop/hive/llap/LlapBaseInputFormat.java b/llap-ext-client/src/java/org/apache/hadoop/hive/llap/LlapBaseInputFormat.java
index 201f5fa..f2d9074 100644
--- a/llap-ext-client/src/java/org/apache/hadoop/hive/llap/LlapBaseInputFormat.java
+++ b/llap-ext-client/src/java/org/apache/hadoop/hive/llap/LlapBaseInputFormat.java
@@ -45,8 +45,8 @@ import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.SubmitWor
import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.VertexOrBinary;
import org.apache.hadoop.hive.llap.ext.LlapTaskUmbilicalExternalClient;
import org.apache.hadoop.hive.llap.ext.LlapTaskUmbilicalExternalClient.LlapTaskUmbilicalExternalResponder;
-import org.apache.hadoop.hive.llap.registry.ServiceInstance;
-import org.apache.hadoop.hive.llap.registry.ServiceInstanceSet;
+import org.apache.hadoop.hive.llap.registry.LlapServiceInstance;
+import org.apache.hadoop.hive.llap.registry.LlapServiceInstanceSet;
import org.apache.hadoop.hive.llap.registry.impl.LlapRegistryService;
import org.apache.hadoop.hive.llap.security.LlapTokenIdentifier;
import org.apache.hadoop.hive.llap.tez.Converters;
@@ -104,7 +104,7 @@ public class LlapBaseInputFormat<V extends WritableComparable<?>>
public static final String PWD_KEY = "llap.if.pwd";
public final String SPLIT_QUERY = "select get_splits(\"%s\",%d)";
- public static final ServiceInstance[] serviceInstanceArray = new ServiceInstance[0];
+ public static final LlapServiceInstance[] serviceInstanceArray = new LlapServiceInstance[0];
public LlapBaseInputFormat(String url, String user, String pwd, String query) {
this.url = url;
@@ -126,7 +126,7 @@ public class LlapBaseInputFormat<V extends WritableComparable<?>>
HiveConf.setVar(job, HiveConf.ConfVars.LLAP_ZK_REGISTRY_USER, llapSplit.getLlapUser());
SubmitWorkInfo submitWorkInfo = SubmitWorkInfo.fromBytes(llapSplit.getPlanBytes());
- ServiceInstance serviceInstance = getServiceInstance(job, llapSplit);
+ LlapServiceInstance serviceInstance = getServiceInstance(job, llapSplit);
String host = serviceInstance.getHost();
int llapSubmitPort = serviceInstance.getRpcPort();
@@ -230,11 +230,11 @@ public class LlapBaseInputFormat<V extends WritableComparable<?>>
return ins.toArray(new InputSplit[ins.size()]);
}
- private ServiceInstance getServiceInstance(JobConf job, LlapInputSplit llapSplit) throws IOException {
+ private LlapServiceInstance getServiceInstance(JobConf job, LlapInputSplit llapSplit) throws IOException {
LlapRegistryService registryService = LlapRegistryService.getClient(job);
String host = llapSplit.getLocations()[0];
- ServiceInstance serviceInstance = getServiceInstanceForHost(registryService, host);
+ LlapServiceInstance serviceInstance = getServiceInstanceForHost(registryService, host);
if (serviceInstance == null) {
LOG.info("No service instances found for " + host + " in registry.");
serviceInstance = getServiceInstanceRandom(registryService);
@@ -246,10 +246,10 @@ public class LlapBaseInputFormat<V extends WritableComparable<?>>
return serviceInstance;
}
- private ServiceInstance getServiceInstanceForHost(LlapRegistryService registryService, String host) throws IOException {
+ private LlapServiceInstance getServiceInstanceForHost(LlapRegistryService registryService, String host) throws IOException {
InetAddress address = InetAddress.getByName(host);
- ServiceInstanceSet instanceSet = registryService.getInstances();
- ServiceInstance serviceInstance = null;
+ LlapServiceInstanceSet instanceSet = registryService.getInstances();
+ LlapServiceInstance serviceInstance = null;
// The name used in the service registry may not match the host name we're using.
// Try hostname/canonical hostname/host address
@@ -279,12 +279,12 @@ public class LlapBaseInputFormat<V extends WritableComparable<?>>
}
- private ServiceInstance getServiceInstanceRandom(LlapRegistryService registryService) throws IOException {
- ServiceInstanceSet instanceSet = registryService.getInstances();
- ServiceInstance serviceInstance = null;
+ private LlapServiceInstance getServiceInstanceRandom(LlapRegistryService registryService) throws IOException {
+ LlapServiceInstanceSet instanceSet = registryService.getInstances();
+ LlapServiceInstance serviceInstance = null;
LOG.info("Finding random live service instance");
- Collection<ServiceInstance> allInstances = instanceSet.getAll();
+ Collection<LlapServiceInstance> allInstances = instanceSet.getAll();
if (allInstances.size() > 0) {
int randIdx = rand.nextInt() % allInstances.size();
serviceInstance = allInstances.toArray(serviceInstanceArray)[randIdx];
@@ -292,13 +292,13 @@ public class LlapBaseInputFormat<V extends WritableComparable<?>>
return serviceInstance;
}
- private ServiceInstance selectServiceInstance(Set<ServiceInstance> serviceInstances) {
+ private LlapServiceInstance selectServiceInstance(Set<LlapServiceInstance> serviceInstances) {
if (serviceInstances == null || serviceInstances.isEmpty()) {
return null;
}
// Get the first live service instance
- for (ServiceInstance serviceInstance : serviceInstances) {
+ for (LlapServiceInstance serviceInstance : serviceInstances) {
return serviceInstance;
}
http://git-wip-us.apache.org/repos/asf/hive/blob/4795b996/llap-server/src/java/org/apache/hadoop/hive/llap/cli/LlapStatusServiceDriver.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/cli/LlapStatusServiceDriver.java b/llap-server/src/java/org/apache/hadoop/hive/llap/cli/LlapStatusServiceDriver.java
index 1b57e38..4fad441 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/cli/LlapStatusServiceDriver.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/cli/LlapStatusServiceDriver.java
@@ -45,7 +45,7 @@ import org.apache.hadoop.hive.llap.cli.status.LlapStatusHelpers;
import org.apache.hadoop.hive.llap.cli.status.LlapStatusHelpers.AppStatusBuilder;
import org.apache.hadoop.hive.llap.cli.status.LlapStatusHelpers.LlapInstance;
import org.apache.hadoop.hive.llap.configuration.LlapDaemonConfiguration;
-import org.apache.hadoop.hive.llap.registry.ServiceInstance;
+import org.apache.hadoop.hive.llap.registry.LlapServiceInstance;
import org.apache.hadoop.hive.llap.registry.impl.LlapRegistryService;
import org.apache.hadoop.hive.ql.session.SessionState;
import org.apache.hadoop.yarn.api.records.ApplicationReport;
@@ -563,7 +563,7 @@ public class LlapStatusServiceDriver {
}
}
- Collection<ServiceInstance> serviceInstances;
+ Collection<LlapServiceInstance> serviceInstances;
try {
serviceInstances = llapRegistry.getInstances(watchTimeoutMs).getAll();
} catch (Exception e) {
@@ -583,7 +583,7 @@ public class LlapStatusServiceDriver {
List<LlapInstance> validatedInstances = new LinkedList<>();
List<String> llapExtraInstances = new LinkedList<>();
- for (ServiceInstance serviceInstance : serviceInstances) {
+ for (LlapServiceInstance serviceInstance : serviceInstances) {
String containerIdString = serviceInstance.getProperties().get(
HiveConf.ConfVars.LLAP_DAEMON_CONTAINER_ID.varname);
http://git-wip-us.apache.org/repos/asf/hive/blob/4795b996/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/services/impl/LlapWebServices.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/services/impl/LlapWebServices.java b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/services/impl/LlapWebServices.java
index ebc3437..58bf8dc 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/services/impl/LlapWebServices.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/services/impl/LlapWebServices.java
@@ -33,7 +33,7 @@ import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.common.classification.InterfaceAudience;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
-import org.apache.hadoop.hive.llap.registry.ServiceInstance;
+import org.apache.hadoop.hive.llap.registry.LlapServiceInstance;
import org.apache.hadoop.hive.llap.registry.impl.LlapRegistryService;
import org.apache.hadoop.hive.llap.registry.impl.LlapZookeeperRegistryImpl;
import org.apache.hadoop.security.UserGroupInformation;
@@ -230,7 +230,7 @@ public class LlapWebServices extends AbstractService {
}
jg.writeStringField("identity", registry.getWorkerIdentity());
jg.writeArrayFieldStart("peers");
- for (ServiceInstance s : registry.getInstances().getAllInstancesOrdered(false)) {
+ for (LlapServiceInstance s : registry.getInstances().getAllInstancesOrdered(false)) {
jg.writeStartObject();
jg.writeStringField("identity", s.getWorkerIdentity());
jg.writeStringField("host", s.getHost());
http://git-wip-us.apache.org/repos/asf/hive/blob/4795b996/llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/LlapTaskCommunicator.java
----------------------------------------------------------------------
diff --git a/llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/LlapTaskCommunicator.java b/llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/LlapTaskCommunicator.java
index ff00aba..100c132 100644
--- a/llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/LlapTaskCommunicator.java
+++ b/llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/LlapTaskCommunicator.java
@@ -15,7 +15,7 @@
package org.apache.hadoop.hive.llap.tezplugins;
import org.apache.commons.lang3.StringUtils;
-import org.apache.hadoop.hive.llap.registry.ServiceInstance;
+import org.apache.hadoop.hive.llap.registry.LlapServiceInstance;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.hive.llap.protocol.LlapTaskUmbilicalProtocol.TezAttemptArray;
@@ -542,7 +542,7 @@ public class LlapTaskCommunicator extends TezTaskCommunicatorImpl {
if (timelineServerUri == null || containerNodeId == null) {
return null;
}
- Set<ServiceInstance> instanceSet;
+ Set<LlapServiceInstance> instanceSet;
try {
instanceSet = serviceRegistry.getInstances().getByHost(containerNodeId.getHost());
} catch (IOException e) {
@@ -554,8 +554,8 @@ public class LlapTaskCommunicator extends TezTaskCommunicatorImpl {
}
// Once NodeId includes fragmentId - this becomes a lot more reliable.
if (instanceSet != null) {
- ServiceInstance matchedInstance = null;
- for (ServiceInstance instance : instanceSet) {
+ LlapServiceInstance matchedInstance = null;
+ for (LlapServiceInstance instance : instanceSet) {
if (instance.getRpcPort() == containerNodeId.getPort()) {
matchedInstance = instance;
break;
http://git-wip-us.apache.org/repos/asf/hive/blob/4795b996/llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/LlapTaskSchedulerService.java
----------------------------------------------------------------------
diff --git a/llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/LlapTaskSchedulerService.java b/llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/LlapTaskSchedulerService.java
index 6bedccb..94f1369 100644
--- a/llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/LlapTaskSchedulerService.java
+++ b/llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/LlapTaskSchedulerService.java
@@ -14,6 +14,8 @@
package org.apache.hadoop.hive.llap.tezplugins;
+import org.apache.hadoop.hive.registry.ServiceInstanceStateChangeListener;
+
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
@@ -62,9 +64,8 @@ import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
import org.apache.hadoop.hive.llap.metrics.LlapMetricsSystem;
import org.apache.hadoop.hive.llap.metrics.MetricsUtils;
-import org.apache.hadoop.hive.llap.registry.ServiceInstance;
-import org.apache.hadoop.hive.llap.registry.ServiceInstanceSet;
-import org.apache.hadoop.hive.llap.registry.ServiceInstanceStateChangeListener;
+import org.apache.hadoop.hive.llap.registry.LlapServiceInstance;
+import org.apache.hadoop.hive.llap.registry.LlapServiceInstanceSet;
import org.apache.hadoop.hive.llap.registry.ServiceRegistry;
import org.apache.hadoop.hive.llap.registry.impl.InactiveServiceInstance;
import org.apache.hadoop.hive.llap.registry.impl.LlapRegistryService;
@@ -107,7 +108,7 @@ public class LlapTaskSchedulerService extends TaskScheduler {
private final Configuration conf;
// interface into the registry service
- private ServiceInstanceSet activeInstances;
+ private LlapServiceInstanceSet activeInstances;
// Tracks all instances, including ones which have been disabled in the past.
// LinkedHashMap to provide the same iteration order when selecting a random host.
@@ -330,7 +331,7 @@ public class LlapTaskSchedulerService extends TaskScheduler {
registry.start();
registry.registerStateChangeListener(new NodeStateChangeListener());
activeInstances = registry.getInstances();
- for (ServiceInstance inst : activeInstances.getAll()) {
+ for (LlapServiceInstance inst : activeInstances.getAll()) {
addNode(new NodeInfo(inst, nodeBlacklistConf, clock,
numSchedulableTasksPerNode, metrics), inst);
}
@@ -340,15 +341,16 @@ public class LlapTaskSchedulerService extends TaskScheduler {
}
@VisibleForTesting
- public void setServiceInstanceSet(ServiceInstanceSet serviceInstanceSet) {
+ public void setServiceInstanceSet(LlapServiceInstanceSet serviceInstanceSet) {
this.activeInstances = serviceInstanceSet;
}
- private class NodeStateChangeListener implements ServiceInstanceStateChangeListener {
+ private class NodeStateChangeListener
+ implements ServiceInstanceStateChangeListener<LlapServiceInstance> {
private final Logger LOG = LoggerFactory.getLogger(NodeStateChangeListener.class);
@Override
- public void onCreate(ServiceInstance serviceInstance) {
+ public void onCreate(LlapServiceInstance serviceInstance) {
LOG.info("Added node with identity: {} as a result of registry callback",
serviceInstance.getWorkerIdentity());
addNode(new NodeInfo(serviceInstance, nodeBlacklistConf, clock,
@@ -356,7 +358,7 @@ public class LlapTaskSchedulerService extends TaskScheduler {
}
@Override
- public void onUpdate(ServiceInstance serviceInstance) {
+ public void onUpdate(LlapServiceInstance serviceInstance) {
// TODO In what situations will this be invoked?
LOG.warn(
"Not expecing Updates from the registry. Received update for instance={}. Ignoring",
@@ -364,7 +366,7 @@ public class LlapTaskSchedulerService extends TaskScheduler {
}
@Override
- public void onRemove(ServiceInstance serviceInstance) {
+ public void onRemove(LlapServiceInstance serviceInstance) {
NodeReport nodeReport = constructNodeReport(serviceInstance, false);
LOG.info("Sending out nodeReport for onRemove: {}", nodeReport);
getContext().nodesUpdated(Collections.singletonList(nodeReport));
@@ -475,7 +477,7 @@ public class LlapTaskSchedulerService extends TaskScheduler {
readLock.lock();
try {
int numInstancesFound = 0;
- for (ServiceInstance inst : activeInstances.getAll()) {
+ for (LlapServiceInstance inst : activeInstances.getAll()) {
Resource r = inst.getResource();
memory += r.getMemory();
vcores += r.getVirtualCores();
@@ -506,7 +508,7 @@ public class LlapTaskSchedulerService extends TaskScheduler {
readLock.lock();
try {
int numInstancesFound = 0;
- for (ServiceInstance inst : activeInstances.getAll()) {
+ for (LlapServiceInstance inst : activeInstances.getAll()) {
NodeInfo nodeInfo = instanceToNodeMap.get(inst.getWorkerIdentity());
if (nodeInfo != null && !nodeInfo.isDisabled()) {
Resource r = inst.getResource();
@@ -766,9 +768,9 @@ public class LlapTaskSchedulerService extends TaskScheduler {
for (String host : requestedHosts) {
prefHostCount++;
// Pick the first host always. Weak attempt at cache affinity.
- Set<ServiceInstance> instances = activeInstances.getByHost(host);
+ Set<LlapServiceInstance> instances = activeInstances.getByHost(host);
if (!instances.isEmpty()) {
- for (ServiceInstance inst : instances) {
+ for (LlapServiceInstance inst : instances) {
NodeInfo nodeInfo = instanceToNodeMap.get(inst.getWorkerIdentity());
if (nodeInfo != null) {
if (nodeInfo.canAcceptTask()) {
@@ -828,10 +830,10 @@ public class LlapTaskSchedulerService extends TaskScheduler {
}
/* fall through - miss in locality or no locality-requested */
- Collection<ServiceInstance> instances = activeInstances.getAllInstancesOrdered(true);
+ Collection<LlapServiceInstance> instances = activeInstances.getAllInstancesOrdered(true);
List<NodeInfo> allNodes = new ArrayList<>(instances.size());
List<NodeInfo> activeNodesWithFreeSlots = new ArrayList<>();
- for (ServiceInstance inst : instances) {
+ for (LlapServiceInstance inst : instances) {
if (inst instanceof InactiveServiceInstance) {
allNodes.add(null);
} else {
@@ -918,7 +920,7 @@ public class LlapTaskSchedulerService extends TaskScheduler {
return new SelectHostResult(randomNode);
}
- private void addNode(NodeInfo node, ServiceInstance serviceInstance) {
+ private void addNode(NodeInfo node, LlapServiceInstance serviceInstance) {
// we have just added a new node. Signal timeout monitor to reset timer
if (activeInstances.size() != 0 && timeoutFutureRef.get() != null) {
LOG.info("New node added. Signalling scheduler timeout monitor thread to stop timer.");
@@ -1006,7 +1008,7 @@ public class LlapTaskSchedulerService extends TaskScheduler {
}
}
- private static NodeReport constructNodeReport(ServiceInstance serviceInstance,
+ private static NodeReport constructNodeReport(LlapServiceInstance serviceInstance,
boolean healthy) {
NodeReport nodeReport = NodeReport.newInstance(NodeId
.newInstance(serviceInstance.getHost(), serviceInstance.getRpcPort()),
@@ -1576,7 +1578,7 @@ public class LlapTaskSchedulerService extends TaskScheduler {
@VisibleForTesting
static class NodeInfo implements Delayed {
private final NodeBlacklistConf blacklistConf;
- final ServiceInstance serviceInstance;
+ final LlapServiceInstance serviceInstance;
private final Clock clock;
long expireTimeMillis = -1;
@@ -1609,7 +1611,7 @@ public class LlapTaskSchedulerService extends TaskScheduler {
* detect based on the serviceInstance, -1 indicates indicates
* @param metrics
*/
- NodeInfo(ServiceInstance serviceInstance, NodeBlacklistConf blacklistConf, Clock clock,
+ NodeInfo(LlapServiceInstance serviceInstance, NodeBlacklistConf blacklistConf, Clock clock,
int numSchedulableTasksConf, final LlapTaskSchedulerMetrics metrics) {
Preconditions.checkArgument(numSchedulableTasksConf >= -1, "NumSchedulableTasks must be >=-1");
this.serviceInstance = serviceInstance;
http://git-wip-us.apache.org/repos/asf/hive/blob/4795b996/llap-tez/src/test/org/apache/hadoop/hive/llap/tezplugins/TestLlapTaskSchedulerService.java
----------------------------------------------------------------------
diff --git a/llap-tez/src/test/org/apache/hadoop/hive/llap/tezplugins/TestLlapTaskSchedulerService.java b/llap-tez/src/test/org/apache/hadoop/hive/llap/tezplugins/TestLlapTaskSchedulerService.java
index 339f513..21a39c7 100644
--- a/llap-tez/src/test/org/apache/hadoop/hive/llap/tezplugins/TestLlapTaskSchedulerService.java
+++ b/llap-tez/src/test/org/apache/hadoop/hive/llap/tezplugins/TestLlapTaskSchedulerService.java
@@ -40,8 +40,8 @@ import java.util.concurrent.locks.ReentrantLock;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
-import org.apache.hadoop.hive.llap.registry.ServiceInstance;
-import org.apache.hadoop.hive.llap.registry.ServiceInstanceSet;
+import org.apache.hadoop.hive.llap.registry.LlapServiceInstance;
+import org.apache.hadoop.hive.llap.registry.LlapServiceInstanceSet;
import org.apache.hadoop.hive.llap.registry.impl.InactiveServiceInstance;
import org.apache.hadoop.hive.llap.registry.impl.LlapFixedRegistryImpl;
import org.apache.hadoop.hive.llap.testhelpers.ControlledClock;
@@ -1447,7 +1447,7 @@ public class TestLlapTaskSchedulerService {
static final Resource resource = Resource.newInstance(1024, 1);
Configuration conf;
TaskSchedulerContext mockAppCallback = mock(TaskSchedulerContext.class);
- ServiceInstanceSet mockServiceInstanceSet = mock(ServiceInstanceSet.class);
+ LlapServiceInstanceSet mockServiceInstanceSet = mock(LlapServiceInstanceSet.class);
ControlledClock clock = new ControlledClock(new MonotonicClock());
ApplicationAttemptId appAttemptId = ApplicationAttemptId.newInstance(ApplicationId.newInstance(1000, 1), 1);
LlapTaskSchedulerServiceForTest ts;
@@ -1499,16 +1499,16 @@ public class TestLlapTaskSchedulerService {
doReturn(userPayload).when(mockAppCallback).getInitialUserPayload();
if (useMockRegistry) {
- List<ServiceInstance> liveInstances = new ArrayList<>();
+ List<LlapServiceInstance> liveInstances = new ArrayList<>();
for (String host : liveHosts) {
if (host == null) {
- ServiceInstance mockInactive = mock(InactiveServiceInstance.class);
+ LlapServiceInstance mockInactive = mock(InactiveServiceInstance.class);
doReturn(host).when(mockInactive).getHost();
doReturn("inactive-host-" + host).when(mockInactive).getWorkerIdentity();
doReturn(ImmutableSet.builder().add(mockInactive).build()).when(mockServiceInstanceSet).getByHost(host);
liveInstances.add(mockInactive);
} else {
- ServiceInstance mockActive = mock(ServiceInstance.class);
+ LlapServiceInstance mockActive = mock(LlapServiceInstance.class);
doReturn(host).when(mockActive).getHost();
doReturn("host-" + host).when(mockActive).getWorkerIdentity();
doReturn(ImmutableSet.builder().add(mockActive).build()).when(mockServiceInstanceSet).getByHost(host);
@@ -1517,9 +1517,9 @@ public class TestLlapTaskSchedulerService {
}
doReturn(liveInstances).when(mockServiceInstanceSet).getAllInstancesOrdered(true);
- List<ServiceInstance> allInstances = new ArrayList<>();
+ List<LlapServiceInstance> allInstances = new ArrayList<>();
for (String host : hosts) {
- ServiceInstance mockActive = mock(ServiceInstance.class);
+ LlapServiceInstance mockActive = mock(LlapServiceInstance.class);
doReturn(host).when(mockActive).getHost();
doReturn(Resource.newInstance(100, 1)).when(mockActive).getResource();
doReturn("host-" + host).when(mockActive).getWorkerIdentity();
http://git-wip-us.apache.org/repos/asf/hive/blob/4795b996/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/Utils.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/Utils.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/Utils.java
index 2b57d90..4c8e7bb 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/Utils.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/Utils.java
@@ -25,7 +25,7 @@ import com.google.common.base.Preconditions;
import org.apache.commons.lang.ArrayUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.conf.HiveConf;
-import org.apache.hadoop.hive.llap.registry.ServiceInstance;
+import org.apache.hadoop.hive.llap.registry.LlapServiceInstance;
import org.apache.hadoop.hive.llap.registry.impl.LlapRegistryService;
import org.apache.hadoop.mapred.InputSplit;
import org.apache.hadoop.mapred.split.SplitLocationProvider;
@@ -43,12 +43,12 @@ public class Utils {
LlapRegistryService serviceRegistry = LlapRegistryService.getClient(conf);
LOG.info("Using LLAP instance " + serviceRegistry.getApplicationId());
- Collection<ServiceInstance> serviceInstances =
+ Collection<LlapServiceInstance> serviceInstances =
serviceRegistry.getInstances().getAllInstancesOrdered(true);
Preconditions.checkArgument(!serviceInstances.isEmpty(),
"No running LLAP daemons! Please check LLAP service status and zookeeper configuration");
ArrayList<String> locations = new ArrayList<>(serviceInstances.size());
- for (ServiceInstance serviceInstance : serviceInstances) {
+ for (LlapServiceInstance serviceInstance : serviceInstances) {
if (LOG.isDebugEnabled()) {
LOG.debug("Adding " + serviceInstance.getWorkerIdentity() + " with hostname=" +
serviceInstance.getHost() + " to list for split locations");
http://git-wip-us.apache.org/repos/asf/hive/blob/4795b996/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/LlapClusterStateForCompile.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/LlapClusterStateForCompile.java b/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/LlapClusterStateForCompile.java
index a5ed308..7a02a56 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/LlapClusterStateForCompile.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/LlapClusterStateForCompile.java
@@ -28,8 +28,8 @@ import java.util.concurrent.Callable;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
-import org.apache.hadoop.hive.llap.registry.ServiceInstance;
-import org.apache.hadoop.hive.llap.registry.ServiceInstanceSet;
+import org.apache.hadoop.hive.llap.registry.LlapServiceInstance;
+import org.apache.hadoop.hive.llap.registry.LlapServiceInstanceSet;
import org.apache.hadoop.hive.llap.registry.impl.InactiveServiceInstance;
import org.apache.hadoop.hive.llap.registry.impl.LlapRegistryService;
import org.slf4j.Logger;
@@ -100,7 +100,7 @@ public class LlapClusterStateForCompile {
return; // Don't fail; this is best-effort.
}
}
- ServiceInstanceSet instances;
+ LlapServiceInstanceSet instances;
try {
instances = svc.getInstances(10);
} catch (IOException e) {
@@ -108,7 +108,7 @@ public class LlapClusterStateForCompile {
return; // Don't wait for the cluster if not started; this is best-effort.
}
int executorsLocal = 0, noConfigNodesLocal = 0;
- for (ServiceInstance si : instances.getAll()) {
+ for (LlapServiceInstance si : instances.getAll()) {
if (si instanceof InactiveServiceInstance) continue; // Shouldn't happen in getAll.
Map<String, String> props = si.getProperties();
if (props == null) {
[2/2] hive git commit: HIVE-17409 : refactor LLAP ZK registry to make
the ZK-registry part reusable (Sergey Shelukhin,
reviewed by Prasanth Jayachandran)
Posted by se...@apache.org.
HIVE-17409 : refactor LLAP ZK registry to make the ZK-registry part reusable (Sergey Shelukhin, reviewed by Prasanth Jayachandran)
Project: http://git-wip-us.apache.org/repos/asf/hive/repo
Commit: http://git-wip-us.apache.org/repos/asf/hive/commit/4795b996
Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/4795b996
Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/4795b996
Branch: refs/heads/master
Commit: 4795b99690c9428636a5a28c5d4be40ce5fc430d
Parents: 2e226d2
Author: sergey <se...@apache.org>
Authored: Fri Sep 1 11:44:11 2017 -0700
Committer: sergey <se...@apache.org>
Committed: Fri Sep 1 11:48:01 2017 -0700
----------------------------------------------------------------------
.../hive/llap/registry/LlapServiceInstance.java | 59 ++
.../llap/registry/LlapServiceInstanceSet.java | 34 +
.../hive/llap/registry/ServiceInstance.java | 86 --
.../hive/llap/registry/ServiceInstanceSet.java | 69 --
.../ServiceInstanceStateChangeListener.java | 42 -
.../hive/llap/registry/ServiceRegistry.java | 8 +-
.../registry/impl/InactiveServiceInstance.java | 4 +-
.../registry/impl/LlapFixedRegistryImpl.java | 41 +-
.../llap/registry/impl/LlapRegistryService.java | 18 +-
.../impl/LlapZookeeperRegistryImpl.java | 800 +++----------------
.../hive/llap/security/LlapTokenClient.java | 18 +-
.../hadoop/hive/registry/ServiceInstance.java | 47 ++
.../hive/registry/ServiceInstanceSet.java | 61 ++
.../ServiceInstanceStateChangeListener.java | 42 +
.../hive/registry/impl/ServiceInstanceBase.java | 93 +++
.../hive/registry/impl/ZkRegistryBase.java | 549 +++++++++++++
.../hive/registry/impl/ZookeeperUtils.java | 116 +++
.../hadoop/hive/llap/LlapBaseInputFormat.java | 30 +-
.../hive/llap/cli/LlapStatusServiceDriver.java | 6 +-
.../daemon/services/impl/LlapWebServices.java | 4 +-
.../llap/tezplugins/LlapTaskCommunicator.java | 8 +-
.../tezplugins/LlapTaskSchedulerService.java | 42 +-
.../TestLlapTaskSchedulerService.java | 16 +-
.../apache/hadoop/hive/ql/exec/tez/Utils.java | 6 +-
.../physical/LlapClusterStateForCompile.java | 8 +-
25 files changed, 1242 insertions(+), 965 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hive/blob/4795b996/llap-client/src/java/org/apache/hadoop/hive/llap/registry/LlapServiceInstance.java
----------------------------------------------------------------------
diff --git a/llap-client/src/java/org/apache/hadoop/hive/llap/registry/LlapServiceInstance.java b/llap-client/src/java/org/apache/hadoop/hive/llap/registry/LlapServiceInstance.java
new file mode 100644
index 0000000..30b1810
--- /dev/null
+++ b/llap-client/src/java/org/apache/hadoop/hive/llap/registry/LlapServiceInstance.java
@@ -0,0 +1,59 @@
+/*
+ * Licensed 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.llap.registry;
+
+
+import org.apache.hadoop.hive.registry.ServiceInstance;
+
+import org.apache.hadoop.yarn.api.records.Resource;
+
+public interface LlapServiceInstance extends ServiceInstance {
+
+ /**
+ * Management endpoint for service instance
+ *
+ * @return
+ */
+ public int getManagementPort();
+
+ /**
+ * Shuffle Endpoint for service instance
+ *
+ * @return
+ */
+ public int getShufflePort();
+
+
+ /**
+ * Address for services hosted on http
+ * @return
+ */
+ public String getServicesAddress();
+ /**
+ * OutputFormat endpoint for service instance
+ *
+ * @return
+ */
+ public int getOutputFormatPort();
+
+
+ /**
+ * Memory and Executors available for the LLAP tasks
+ *
+ * This does not include the size of the cache or the actual vCores allocated via Slider.
+ *
+ * @return
+ */
+ public Resource getResource();
+}
http://git-wip-us.apache.org/repos/asf/hive/blob/4795b996/llap-client/src/java/org/apache/hadoop/hive/llap/registry/LlapServiceInstanceSet.java
----------------------------------------------------------------------
diff --git a/llap-client/src/java/org/apache/hadoop/hive/llap/registry/LlapServiceInstanceSet.java b/llap-client/src/java/org/apache/hadoop/hive/llap/registry/LlapServiceInstanceSet.java
new file mode 100644
index 0000000..a728f4a
--- /dev/null
+++ b/llap-client/src/java/org/apache/hadoop/hive/llap/registry/LlapServiceInstanceSet.java
@@ -0,0 +1,34 @@
+/*
+ * Licensed 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.llap.registry;
+
+import java.util.Collection;
+import org.apache.hadoop.hive.registry.ServiceInstanceSet;
+import org.apache.hadoop.yarn.api.records.ApplicationId;
+
+public interface LlapServiceInstanceSet extends ServiceInstanceSet<LlapServiceInstance> {
+
+ /**
+ * Gets a list containing all the instances. This list has the same iteration order across
+ * different processes, assuming the list of registry entries is the same.
+ * @param consistentIndexes if true, also try to maintain the same exact index for each node
+ * across calls, by inserting inactive instances to replace the
+ * removed ones.
+ */
+ Collection<LlapServiceInstance> getAllInstancesOrdered(
+ boolean consistentIndexes);
+
+ /** LLAP application ID */
+ ApplicationId getApplicationId();
+}
http://git-wip-us.apache.org/repos/asf/hive/blob/4795b996/llap-client/src/java/org/apache/hadoop/hive/llap/registry/ServiceInstance.java
----------------------------------------------------------------------
diff --git a/llap-client/src/java/org/apache/hadoop/hive/llap/registry/ServiceInstance.java b/llap-client/src/java/org/apache/hadoop/hive/llap/registry/ServiceInstance.java
deleted file mode 100644
index 70515c4..0000000
--- a/llap-client/src/java/org/apache/hadoop/hive/llap/registry/ServiceInstance.java
+++ /dev/null
@@ -1,86 +0,0 @@
-/*
- * Licensed 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.llap.registry;
-
-import java.util.Map;
-
-import org.apache.hadoop.yarn.api.records.Resource;
-
-public interface ServiceInstance {
-
- /**
- * Worker identity is a UUID (unique across restarts), to identify a node which died & was brought
- * back on the same host/port
- */
- public String getWorkerIdentity();
-
- /**
- * Hostname of the service instance
- *
- * @return
- */
- public String getHost();
-
- /**
- * RPC Endpoint for service instance
- *
- * @return
- */
- public int getRpcPort();
-
- /**
- * Management endpoint for service instance
- *
- * @return
- */
- public int getManagementPort();
-
- /**
- * Shuffle Endpoint for service instance
- *
- * @return
- */
- public int getShufflePort();
-
-
- /**
- * Address for services hosted on http
- * @return
- */
- public String getServicesAddress();
- /**
- * OutputFormat endpoint for service instance
- *
- * @return
- */
- public int getOutputFormatPort();
-
-
- /**
- * Config properties of the Service Instance (llap.daemon.*)
- *
- * @return
- */
-
- public Map<String, String> getProperties();
-
- /**
- * Memory and Executors available for the LLAP tasks
- *
- * This does not include the size of the cache or the actual vCores allocated via Slider.
- *
- * @return
- */
- public Resource getResource();
-}
http://git-wip-us.apache.org/repos/asf/hive/blob/4795b996/llap-client/src/java/org/apache/hadoop/hive/llap/registry/ServiceInstanceSet.java
----------------------------------------------------------------------
diff --git a/llap-client/src/java/org/apache/hadoop/hive/llap/registry/ServiceInstanceSet.java b/llap-client/src/java/org/apache/hadoop/hive/llap/registry/ServiceInstanceSet.java
deleted file mode 100644
index cc124e7..0000000
--- a/llap-client/src/java/org/apache/hadoop/hive/llap/registry/ServiceInstanceSet.java
+++ /dev/null
@@ -1,69 +0,0 @@
-/*
- * Licensed 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.llap.registry;
-
-import java.util.Collection;
-import java.util.Set;
-
-/**
- * Note: For most of the implementations, there's no guarantee that the ServiceInstance returned by
- * one invocation is the same as the instance returned by another invocation. e.g. the ZK registry
- * returns a new ServiceInstance object each time a getInstance call is made.
- */
-public interface ServiceInstanceSet {
-
- /**
- * Get an instance mapping which map worker identity to each instance.
- *
- * The worker identity does not collide between restarts, so each restart will have a unique id,
- * while having the same host/ip pair.
- *
- * @return
- */
- Collection<ServiceInstance> getAll();
-
- /**
- * Gets a list containing all the instances. This list has the same iteration order across
- * different processes, assuming the list of registry entries is the same.
- * @param consistentIndexes if true, also try to maintain the same exact index for each node
- * across calls, by inserting inactive instances to replace the
- * removed ones.
- */
- Collection<ServiceInstance> getAllInstancesOrdered(boolean consistentIndexes);
-
- /**
- * Get an instance by worker identity.
- *
- * @param name
- * @return
- */
- ServiceInstance getInstance(String name);
-
- /**
- * Get a list of service instances for a given host.
- *
- * The list could include dead and alive instances.
- *
- * @param host
- * @return
- */
- Set<ServiceInstance> getByHost(String host);
-
- /**
- * Get number of instances in the currently availabe.
- *
- * @return - number of instances
- */
- int size();
-}
http://git-wip-us.apache.org/repos/asf/hive/blob/4795b996/llap-client/src/java/org/apache/hadoop/hive/llap/registry/ServiceInstanceStateChangeListener.java
----------------------------------------------------------------------
diff --git a/llap-client/src/java/org/apache/hadoop/hive/llap/registry/ServiceInstanceStateChangeListener.java b/llap-client/src/java/org/apache/hadoop/hive/llap/registry/ServiceInstanceStateChangeListener.java
deleted file mode 100644
index 92eb8bd..0000000
--- a/llap-client/src/java/org/apache/hadoop/hive/llap/registry/ServiceInstanceStateChangeListener.java
+++ /dev/null
@@ -1,42 +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.llap.registry;
-
-/**
- * Callback listener for instance state change events
- */
-public interface ServiceInstanceStateChangeListener {
- /**
- * Called when new {@link ServiceInstance} is created.
- *
- * @param serviceInstance - created service instance
- */
- void onCreate(ServiceInstance serviceInstance);
-
- /**
- * Called when an existing {@link ServiceInstance} is updated.
- *
- * @param serviceInstance - updated service instance
- */
- void onUpdate(ServiceInstance serviceInstance);
-
- /**
- * Called when an existing {@link ServiceInstance} is removed.
- *
- * @param serviceInstance - removed service instance
- */
- void onRemove(ServiceInstance serviceInstance);
-}
http://git-wip-us.apache.org/repos/asf/hive/blob/4795b996/llap-client/src/java/org/apache/hadoop/hive/llap/registry/ServiceRegistry.java
----------------------------------------------------------------------
diff --git a/llap-client/src/java/org/apache/hadoop/hive/llap/registry/ServiceRegistry.java b/llap-client/src/java/org/apache/hadoop/hive/llap/registry/ServiceRegistry.java
index 5739d72..5d7f813 100644
--- a/llap-client/src/java/org/apache/hadoop/hive/llap/registry/ServiceRegistry.java
+++ b/llap-client/src/java/org/apache/hadoop/hive/llap/registry/ServiceRegistry.java
@@ -14,7 +14,7 @@
package org.apache.hadoop.hive.llap.registry;
import java.io.IOException;
-
+import org.apache.hadoop.hive.registry.ServiceInstanceStateChangeListener;
import org.apache.hadoop.yarn.api.records.ApplicationId;
/**
@@ -49,14 +49,14 @@ public interface ServiceRegistry {
* @param clusterReadyTimeoutMs The time to wait for the cluster to be ready, if it's not
* started yet. 0 means do not wait.
*/
- ServiceInstanceSet getInstances(String component, long clusterReadyTimeoutMs) throws IOException;
+ LlapServiceInstanceSet getInstances(String component, long clusterReadyTimeoutMs) throws IOException;
/**
* Adds state change listeners for service instances.
* @param listener - state change listener
*/
- void registerStateChangeListener(ServiceInstanceStateChangeListener listener)
- throws IOException;
+ void registerStateChangeListener(
+ ServiceInstanceStateChangeListener<LlapServiceInstance> listener) throws IOException;
/**
* @return The application ID of the LLAP cluster.
http://git-wip-us.apache.org/repos/asf/hive/blob/4795b996/llap-client/src/java/org/apache/hadoop/hive/llap/registry/impl/InactiveServiceInstance.java
----------------------------------------------------------------------
diff --git a/llap-client/src/java/org/apache/hadoop/hive/llap/registry/impl/InactiveServiceInstance.java b/llap-client/src/java/org/apache/hadoop/hive/llap/registry/impl/InactiveServiceInstance.java
index 9f2f3b4..1d6b716 100644
--- a/llap-client/src/java/org/apache/hadoop/hive/llap/registry/impl/InactiveServiceInstance.java
+++ b/llap-client/src/java/org/apache/hadoop/hive/llap/registry/impl/InactiveServiceInstance.java
@@ -16,10 +16,10 @@ package org.apache.hadoop.hive.llap.registry.impl;
import java.util.Map;
-import org.apache.hadoop.hive.llap.registry.ServiceInstance;
+import org.apache.hadoop.hive.llap.registry.LlapServiceInstance;
import org.apache.hadoop.yarn.api.records.Resource;
-public class InactiveServiceInstance implements ServiceInstance {
+public class InactiveServiceInstance implements LlapServiceInstance {
private final String name;
public InactiveServiceInstance(String name) {
this.name = name;
http://git-wip-us.apache.org/repos/asf/hive/blob/4795b996/llap-client/src/java/org/apache/hadoop/hive/llap/registry/impl/LlapFixedRegistryImpl.java
----------------------------------------------------------------------
diff --git a/llap-client/src/java/org/apache/hadoop/hive/llap/registry/impl/LlapFixedRegistryImpl.java b/llap-client/src/java/org/apache/hadoop/hive/llap/registry/impl/LlapFixedRegistryImpl.java
index ebc32a1..c88198f 100644
--- a/llap-client/src/java/org/apache/hadoop/hive/llap/registry/impl/LlapFixedRegistryImpl.java
+++ b/llap-client/src/java/org/apache/hadoop/hive/llap/registry/impl/LlapFixedRegistryImpl.java
@@ -29,15 +29,14 @@ import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
-
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
-import org.apache.hadoop.hive.llap.registry.ServiceInstance;
-import org.apache.hadoop.hive.llap.registry.ServiceInstanceSet;
-import org.apache.hadoop.hive.llap.registry.ServiceInstanceStateChangeListener;
+import org.apache.hadoop.hive.llap.registry.LlapServiceInstance;
+import org.apache.hadoop.hive.llap.registry.LlapServiceInstanceSet;
import org.apache.hadoop.hive.llap.registry.ServiceRegistry;
+import org.apache.hadoop.hive.registry.ServiceInstanceStateChangeListener;
import org.apache.hadoop.net.NetUtils;
import org.apache.hadoop.util.StringUtils;
import org.apache.hadoop.yarn.api.records.ApplicationId;
@@ -117,7 +116,7 @@ public class LlapFixedRegistryImpl implements ServiceRegistry {
return "host-" + host;
}
- private final class FixedServiceInstance implements ServiceInstance {
+ private final class FixedServiceInstance implements LlapServiceInstance {
private final String host;
private final String serviceAddress;
@@ -206,10 +205,10 @@ public class LlapFixedRegistryImpl implements ServiceRegistry {
}
}
- private final class FixedServiceInstanceSet implements ServiceInstanceSet {
+ private final class FixedServiceInstanceSet implements LlapServiceInstanceSet {
// LinkedHashMap have a repeatable iteration order.
- private final Map<String, ServiceInstance> instances = new LinkedHashMap<>();
+ private final Map<String, LlapServiceInstance> instances = new LinkedHashMap<>();
public FixedServiceInstanceSet() {
for (String host : hosts) {
@@ -219,17 +218,17 @@ public class LlapFixedRegistryImpl implements ServiceRegistry {
}
@Override
- public Collection<ServiceInstance> getAll() {
+ public Collection<LlapServiceInstance> getAll() {
return instances.values();
}
@Override
- public List<ServiceInstance> getAllInstancesOrdered(boolean consistentIndexes) {
- List<ServiceInstance> list = new LinkedList<>();
+ public List<LlapServiceInstance> getAllInstancesOrdered(boolean consistentIndexes) {
+ List<LlapServiceInstance> list = new LinkedList<>();
list.addAll(instances.values());
- Collections.sort(list, new Comparator<ServiceInstance>() {
+ Collections.sort(list, new Comparator<LlapServiceInstance>() {
@Override
- public int compare(ServiceInstance o1, ServiceInstance o2) {
+ public int compare(LlapServiceInstance o1, LlapServiceInstance o2) {
return o2.getWorkerIdentity().compareTo(o2.getWorkerIdentity());
}
});
@@ -237,14 +236,14 @@ public class LlapFixedRegistryImpl implements ServiceRegistry {
}
@Override
- public ServiceInstance getInstance(String name) {
+ public LlapServiceInstance getInstance(String name) {
return instances.get(name);
}
@Override
- public Set<ServiceInstance> getByHost(String host) {
- Set<ServiceInstance> byHost = new HashSet<ServiceInstance>();
- ServiceInstance inst = getInstance(getWorkerIdentity(host));
+ public Set<LlapServiceInstance> getByHost(String host) {
+ Set<LlapServiceInstance> byHost = new HashSet<LlapServiceInstance>();
+ LlapServiceInstance inst = getInstance(getWorkerIdentity(host));
if (inst != null) {
byHost.add(inst);
}
@@ -255,15 +254,21 @@ public class LlapFixedRegistryImpl implements ServiceRegistry {
public int size() {
return instances.size();
}
+
+ @Override
+ public ApplicationId getApplicationId() {
+ return null;
+ }
}
@Override
- public ServiceInstanceSet getInstances(String component, long timeoutMs) throws IOException {
+ public LlapServiceInstanceSet getInstances(String component, long timeoutMs) throws IOException {
return new FixedServiceInstanceSet();
}
@Override
- public void registerStateChangeListener(final ServiceInstanceStateChangeListener listener) {
+ public void registerStateChangeListener(
+ final ServiceInstanceStateChangeListener<LlapServiceInstance> listener) {
// nothing to set
LOG.warn("Callbacks for instance state changes are not supported in fixed registry.");
}
http://git-wip-us.apache.org/repos/asf/hive/blob/4795b996/llap-client/src/java/org/apache/hadoop/hive/llap/registry/impl/LlapRegistryService.java
----------------------------------------------------------------------
diff --git a/llap-client/src/java/org/apache/hadoop/hive/llap/registry/impl/LlapRegistryService.java b/llap-client/src/java/org/apache/hadoop/hive/llap/registry/impl/LlapRegistryService.java
index 76fc9c7..80a6aba 100644
--- a/llap-client/src/java/org/apache/hadoop/hive/llap/registry/impl/LlapRegistryService.java
+++ b/llap-client/src/java/org/apache/hadoop/hive/llap/registry/impl/LlapRegistryService.java
@@ -13,24 +13,24 @@
*/
package org.apache.hadoop.hive.llap.registry.impl;
+import org.apache.hadoop.hive.llap.registry.LlapServiceInstance;
+
+import com.google.common.base.Preconditions;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
-
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
-import org.apache.hadoop.hive.llap.registry.ServiceInstanceSet;
-import org.apache.hadoop.hive.llap.registry.ServiceInstanceStateChangeListener;
+import org.apache.hadoop.hive.llap.registry.LlapServiceInstanceSet;
import org.apache.hadoop.hive.llap.registry.ServiceRegistry;
+import org.apache.hadoop.hive.registry.ServiceInstanceStateChangeListener;
import org.apache.hadoop.registry.client.binding.RegistryUtils;
import org.apache.hadoop.service.AbstractService;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.google.common.base.Preconditions;
-
public class LlapRegistryService extends AbstractService {
private static final Logger LOG = LoggerFactory.getLogger(LlapRegistryService.class);
@@ -131,16 +131,16 @@ public class LlapRegistryService extends AbstractService {
}
}
- public ServiceInstanceSet getInstances() throws IOException {
+ public LlapServiceInstanceSet getInstances() throws IOException {
return getInstances(0);
}
- public ServiceInstanceSet getInstances(long clusterReadyTimeoutMs) throws IOException {
+ public LlapServiceInstanceSet getInstances(long clusterReadyTimeoutMs) throws IOException {
return this.registry.getInstances("LLAP", clusterReadyTimeoutMs);
}
- public void registerStateChangeListener(ServiceInstanceStateChangeListener listener)
- throws IOException {
+ public void registerStateChangeListener(
+ ServiceInstanceStateChangeListener<LlapServiceInstance> listener) throws IOException {
this.registry.registerStateChangeListener(listener);
}
http://git-wip-us.apache.org/repos/asf/hive/blob/4795b996/llap-client/src/java/org/apache/hadoop/hive/llap/registry/impl/LlapZookeeperRegistryImpl.java
----------------------------------------------------------------------
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 ad17144..65f8f94 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
@@ -13,83 +13,47 @@
*/
package org.apache.hadoop.hive.llap.registry.impl;
+import org.apache.hadoop.registry.client.binding.RegistryUtils;
+
+import com.google.common.collect.Sets;
import java.io.IOException;
-import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.MalformedURLException;
import java.net.URISyntaxException;
import java.net.URL;
-import java.net.UnknownHostException;
-import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
-import java.util.HashSet;
import java.util.Iterator;
-import java.util.LinkedList;
-import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
-import java.util.UUID;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantLock;
-
-import javax.security.auth.login.AppConfigurationEntry;
-
-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;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
-import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
-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;
import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
-import org.apache.hadoop.hive.llap.LlapUtil;
import org.apache.hadoop.hive.llap.io.api.LlapProxy;
-import org.apache.hadoop.hive.llap.registry.ServiceInstance;
-import org.apache.hadoop.hive.llap.registry.ServiceInstanceSet;
-import org.apache.hadoop.hive.llap.registry.ServiceInstanceStateChangeListener;
+import org.apache.hadoop.hive.llap.registry.LlapServiceInstance;
+import org.apache.hadoop.hive.llap.registry.LlapServiceInstanceSet;
import org.apache.hadoop.hive.llap.registry.ServiceRegistry;
+import org.apache.hadoop.hive.registry.impl.ServiceInstanceBase;
+import org.apache.hadoop.hive.registry.impl.ZkRegistryBase;
import org.apache.hadoop.registry.client.binding.RegistryTypeUtils;
-import org.apache.hadoop.registry.client.binding.RegistryUtils;
import org.apache.hadoop.registry.client.binding.RegistryUtils.ServiceRecordMarshal;
import org.apache.hadoop.registry.client.types.AddressTypes;
import org.apache.hadoop.registry.client.types.Endpoint;
import org.apache.hadoop.registry.client.types.ProtocolTypes;
import org.apache.hadoop.registry.client.types.ServiceRecord;
-import org.apache.hadoop.security.SecurityUtil;
-import org.apache.hadoop.security.UserGroupInformation;
-import org.apache.hadoop.security.authentication.util.KerberosUtil;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.ContainerId;
import org.apache.hadoop.yarn.api.records.Resource;
-import org.apache.hadoop.yarn.conf.YarnConfiguration;
-import org.apache.zookeeper.ZooDefs;
-import org.apache.zookeeper.KeeperException.InvalidACLException;
-import org.apache.zookeeper.client.ZooKeeperSaslClient;
-import org.apache.zookeeper.data.ACL;
-import org.apache.zookeeper.data.Id;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.google.common.base.Preconditions;
-import com.google.common.collect.Lists;
-import com.google.common.util.concurrent.ThreadFactoryBuilder;
-
-public class LlapZookeeperRegistryImpl implements ServiceRegistry {
-
+public class LlapZookeeperRegistryImpl
+ extends ZkRegistryBase<LlapServiceInstance> implements ServiceRegistry {
private static final Logger LOG = LoggerFactory.getLogger(LlapZookeeperRegistryImpl.class);
/**
@@ -100,155 +64,29 @@ public class LlapZookeeperRegistryImpl implements ServiceRegistry {
private static final String IPC_SHUFFLE = "shuffle";
private static final String IPC_LLAP = "llap";
private static final String IPC_OUTPUTFORMAT = "llapoutputformat";
- private final static String SASL_NAMESPACE = "llap-sasl";
- private final static String UNSECURE_NAMESPACE = "llap-unsecure";
+ private final static String NAMESPACE_PREFIX = "llap-";
private final static String USER_SCOPE_PATH_PREFIX = "user-";
- private static final String DISABLE_MESSAGE =
- "Set " + ConfVars.LLAP_VALIDATE_ACLS.varname + " to false to disable ACL validation";
private static final String WORKER_PREFIX = "worker-";
private static final String SLOT_PREFIX = "slot-";
+ private static final String SASL_LOGIN_CONTEXT_NAME = "LlapZooKeeperClient";
- private final Configuration conf;
- private final CuratorFramework zooKeeperClient;
- // userPathPrefix is the path specific to the user for which ACLs should be restrictive.
- // workersPath is the directory path where all the worker znodes are located.
- private final String userPathPrefix, workersPath;
- private String userNameFromPrincipal; // Only set when setting up the secure config for ZK.
-
- private PersistentEphemeralNode znode;
private SlotZnode slotZnode;
- private String znodePath; // unique identity for this instance
- private final ServiceRecordMarshal encoder; // to marshal/unmarshal znode data
- // to be used by clients of ServiceRegistry
+ // to be used by clients of ServiceRegistry TODO: this is unnecessary
private DynamicServiceInstanceSet instances;
- private PathChildrenCache instancesCache;
-
- private static final UUID uniq = UUID.randomUUID();
- private static final String UNIQUE_IDENTIFIER = "llap.unique.id";
-
- private Set<ServiceInstanceStateChangeListener> stateChangeListeners;
- private final Map<String, Set<ServiceInstance>> pathToInstanceCache;
- private final Map<String, Set<ServiceInstance>> nodeToInstanceCache;
- private final Lock instanceCacheLock = new ReentrantLock();
-
- // get local hostname
- private static final String hostname;
-
- static {
- String localhost = "localhost";
- try {
- localhost = InetAddress.getLocalHost().getCanonicalHostName();
- } catch (UnknownHostException uhe) {
- // ignore
- }
- hostname = localhost;
- }
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
- // worker does not respond due to communication interruptions it will retain the same sequence
- // number when it returns back. If session timeout expires, the node will be deleted and new
- // addition of the same node (restart) will get next sequence number
- this.userPathPrefix = USER_SCOPE_PATH_PREFIX + getZkPathUser(this.conf);
- this.workersPath = "/" + userPathPrefix + "/" + instanceName + "/workers";
- this.instancesCache = null;
- this.instances = null;
- this.stateChangeListeners = new HashSet<>();
- this.pathToInstanceCache = new ConcurrentHashMap<>();
- this.nodeToInstanceCache = new ConcurrentHashMap<>();
-
- final boolean isSecure = UserGroupInformation.isSecurityEnabled();
- ACLProvider zooKeeperAclProvider = new ACLProvider() {
- @Override
- public List<ACL> getDefaultAcl() {
- // We always return something from getAclForPath so this should not happen.
- LOG.warn("getDefaultAcl was called");
- return Lists.newArrayList(ZooDefs.Ids.OPEN_ACL_UNSAFE);
- }
-
- @Override
- public List<ACL> getAclForPath(String path) {
- if (!isSecure || path == null || !path.contains(userPathPrefix)) {
- // No security or the path is below the user path - full access.
- return Lists.newArrayList(ZooDefs.Ids.OPEN_ACL_UNSAFE);
- }
- return createSecureAcls();
- }
- };
- String rootNs = HiveConf.getVar(conf, ConfVars.LLAP_ZK_REGISTRY_NAMESPACE);
- if (rootNs == null) {
- 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();
-
+ super(instanceName, conf,
+ HiveConf.getVar(conf, ConfVars.LLAP_ZK_REGISTRY_NAMESPACE), NAMESPACE_PREFIX,
+ USER_SCOPE_PATH_PREFIX, WORKER_PREFIX,
+ LlapProxy.isDaemon() ? SASL_LOGIN_CONTEXT_NAME : null,
+ HiveConf.getVar(conf, ConfVars.LLAP_KERBEROS_PRINCIPAL),
+ HiveConf.getVar(conf, ConfVars.LLAP_KERBEROS_KEYTAB_FILE),
+ ConfVars.LLAP_VALIDATE_ACLS);
LOG.info("Llap Zookeeper Registry is enabled with registryid: " + instanceName);
}
- private static List<ACL> createSecureAcls() {
- // Read all to the world
- List<ACL> nodeAcls = new ArrayList<ACL>(ZooDefs.Ids.READ_ACL_UNSAFE);
- // Create/Delete/Write/Admin to creator
- nodeAcls.addAll(ZooDefs.Ids.CREATOR_ALL_ACL);
- return nodeAcls;
- }
-
- /**
- * Get the ensemble server addresses from the configuration. The format is: host1:port,
- * host2:port..
- *
- * @param conf
- **/
- private String getQuorumServers(Configuration conf) {
- String[] hosts = conf.getTrimmedStrings(ConfVars.HIVE_ZOOKEEPER_QUORUM.varname);
- String port = conf.get(ConfVars.HIVE_ZOOKEEPER_CLIENT_PORT.varname,
- ConfVars.HIVE_ZOOKEEPER_CLIENT_PORT.getDefaultValue());
- 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();
- }
-
- private String getZkPathUser(Configuration conf) {
- // External LLAP clients would need to set LLAP_ZK_REGISTRY_USER to the LLAP daemon user (hive),
- // rather than relying on RegistryUtils.currentUser().
- String user = HiveConf.getVar(conf, ConfVars.LLAP_ZK_REGISTRY_USER, RegistryUtils.currentUser());
- return user;
- }
-
public Endpoint getRpcEndpoint() {
final int rpcPort = HiveConf.getIntVar(conf, ConfVars.LLAP_DAEMON_RPC_PORT);
return RegistryTypeUtils.ipcEndpoint(IPC_LLAP, new InetSocketAddress(hostname, rpcPort));
@@ -304,117 +142,37 @@ public class LlapZookeeperRegistryImpl implements ServiceRegistry {
}
}
- // restart sensitive instance id
- srv.set(UNIQUE_IDENTIFIER, uniq.toString());
+ String uniqueId = registerServiceRecord(srv);
+ long znodeCreationTimeout = 120;
// Create a znode under the rootNamespace parent for this instance of the server
try {
- // PersistentEphemeralNode will make sure the ephemeral node created on server will be present
- // even under connection or session interruption (will automatically handle retries)
- znode = new PersistentEphemeralNode(zooKeeperClient, Mode.EPHEMERAL_SEQUENTIAL,
- workersPath + "/" + WORKER_PREFIX, encoder.toBytes(srv));
-
- // start the creation of znodes
- 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");
- }
-
- znodePath = znode.getActualPath();
-
slotZnode = new SlotZnode(
- zooKeeperClient, workersPath, SLOT_PREFIX, WORKER_PREFIX, uniq.toString());
+ zooKeeperClient, workersPath, SLOT_PREFIX, WORKER_PREFIX, uniqueId);
if (!slotZnode.start(znodeCreationTimeout, TimeUnit.SECONDS)) {
throw new Exception(
"Max znode creation wait time: " + znodeCreationTimeout + "s exhausted");
}
-
- if (HiveConf.getBoolVar(conf, ConfVars.LLAP_VALIDATE_ACLS)) {
- try {
- checkAndSetAcls();
- } catch (Exception ex) {
- throw new IOException("Error validating or setting ACLs. " + DISABLE_MESSAGE, ex);
- }
- }
- if (zooKeeperClient.checkExists().forPath(znodePath) == null) {
- // No node exists, throw exception
- throw new Exception("Unable to create znode for this LLAP instance on ZooKeeper.");
- }
- LOG.info(
- "Registered node. Created a znode on ZooKeeper for LLAP instance: rpc: {}, shuffle: {}," +
- " webui: {}, mgmt: {}, znodePath: {} ",
- rpcEndpoint, getShuffleEndpoint(), getServicesEndpoint(), getMngEndpoint(), znodePath);
} catch (Exception e) {
LOG.error("Unable to create a znode for this server instance", e);
- CloseableUtils.closeQuietly(znode);
CloseableUtils.closeQuietly(slotZnode);
+ super.stop();
throw (e instanceof IOException) ? (IOException)e : new IOException(e);
}
- if (LOG.isDebugEnabled()) {
- LOG.debug("Created zknode with path: {} service record: {}", znodePath, srv);
- }
-
- return uniq.toString();
+ LOG.info("Registered node. Created a znode on ZooKeeper for LLAP instance: rpc: {}, " +
+ "shuffle: {}, webui: {}, mgmt: {}, znodePath: {}", rpcEndpoint, getShuffleEndpoint(),
+ getServicesEndpoint(), getMngEndpoint(), getRegistrationZnodePath());
+ return uniqueId;
}
- private void checkAndSetAcls() throws Exception {
- if (!UserGroupInformation.isSecurityEnabled()) return;
- // We are trying to check ACLs on the "workers" directory, which noone except us should be
- // able to write to. Higher-level directories shouldn't matter - we don't read them.
- String pathToCheck = workersPath;
- List<ACL> acls = zooKeeperClient.getACL().forPath(pathToCheck);
- if (acls == null || acls.isEmpty()) {
- // Can there be no ACLs? There's some access (to get ACLs), so assume it means free for all.
- LOG.warn("No ACLs on " + pathToCheck + "; setting up ACLs. " + DISABLE_MESSAGE);
- setUpAcls(pathToCheck);
- return;
- }
- // This could be brittle.
- assert userNameFromPrincipal != null;
- Id currentUser = new Id("sasl", userNameFromPrincipal);
- for (ACL acl : acls) {
- if ((acl.getPerms() & ~ZooDefs.Perms.READ) == 0 || currentUser.equals(acl.getId())) {
- continue; // Read permission/no permissions, or the expected user.
- }
- LOG.warn("The ACL " + acl + " is unnacceptable for " + pathToCheck
- + "; setting up ACLs. " + DISABLE_MESSAGE);
- setUpAcls(pathToCheck);
- return;
- }
- }
-
- private void setUpAcls(String path) throws Exception {
- List<ACL> acls = createSecureAcls();
- LinkedList<String> paths = new LinkedList<>();
- paths.add(path);
- while (!paths.isEmpty()) {
- String currentPath = paths.poll();
- List<String> children = zooKeeperClient.getChildren().forPath(currentPath);
- if (children != null) {
- for (String child : children) {
- paths.add(currentPath + "/" + child);
- }
- }
- zooKeeperClient.setACL().withACL(acls).forPath(currentPath);
- }
- }
-
-
@Override
public void unregister() throws IOException {
// Nothing for the zkCreate models
}
- private class DynamicServiceInstance implements ServiceInstance {
-
- private final ServiceRecord srv;
- private final String host;
- private final int rpcPort;
+ private class DynamicServiceInstance
+ extends ServiceInstanceBase implements LlapServiceInstance {
private final int mngPort;
private final int shufflePort;
private final int outputFormatPort;
@@ -422,24 +180,13 @@ public class LlapZookeeperRegistryImpl implements ServiceRegistry {
private final Resource resource;
public DynamicServiceInstance(ServiceRecord srv) throws IOException {
- this.srv = srv;
-
- if (LOG.isTraceEnabled()) {
- LOG.trace("Working with ServiceRecord: {}", srv);
- }
+ super(srv, IPC_LLAP);
final Endpoint shuffle = srv.getInternalEndpoint(IPC_SHUFFLE);
- final Endpoint rpc = srv.getInternalEndpoint(IPC_LLAP);
final Endpoint mng = srv.getInternalEndpoint(IPC_MNG);
final Endpoint outputFormat = srv.getInternalEndpoint(IPC_OUTPUTFORMAT);
final Endpoint services = srv.getExternalEndpoint(IPC_SERVICES);
- this.host =
- RegistryTypeUtils.getAddressField(rpc.addresses.get(0),
- AddressTypes.ADDRESS_HOSTNAME_FIELD);
- this.rpcPort =
- Integer.parseInt(RegistryTypeUtils.getAddressField(rpc.addresses.get(0),
- AddressTypes.ADDRESS_PORT_FIELD));
this.mngPort =
Integer.parseInt(RegistryTypeUtils.getAddressField(mng.addresses.get(0),
AddressTypes.ADDRESS_PORT_FIELD));
@@ -462,39 +209,6 @@ public class LlapZookeeperRegistryImpl implements ServiceRegistry {
}
@Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
-
- DynamicServiceInstance other = (DynamicServiceInstance) o;
- return this.getWorkerIdentity().equals(other.getWorkerIdentity());
- }
-
- @Override
- public int hashCode() {
- return getWorkerIdentity().hashCode();
- }
-
- @Override
- public String getWorkerIdentity() {
- return srv.get(UNIQUE_IDENTIFIER);
- }
-
- @Override
- public String getHost() {
- return host;
- }
-
- @Override
- public int getRpcPort() {
- return rpcPort;
- }
-
- @Override
public int getShufflePort() {
return shufflePort;
}
@@ -505,11 +219,6 @@ public class LlapZookeeperRegistryImpl implements ServiceRegistry {
}
@Override
- public Map<String, String> getProperties() {
- return srv.attributes();
- }
-
- @Override
public Resource getResource() {
return resource;
}
@@ -530,83 +239,60 @@ public class LlapZookeeperRegistryImpl implements ServiceRegistry {
public int getOutputFormatPort() {
return outputFormatPort;
}
-
- // TODO: This needs a hashCode/equality implementation if used as a key in various structures.
- // A new ServiceInstance is created each time.
}
- private void addToCache(String path, String host, ServiceInstance instance) {
- instanceCacheLock.lock();
- try {
- putInCache(path, pathToInstanceCache, instance);
- putInCache(host, nodeToInstanceCache, instance);
- } finally {
- instanceCacheLock.unlock();
- }
- LOG.debug("Added path={}, host={} instance={} to cache."
- + " pathToInstanceCache:size={}, nodeToInstanceCache:size={}",
- path, host, instance, pathToInstanceCache.size(), nodeToInstanceCache.size());
- }
- private void removeFromCache(String path, String host) {
- instanceCacheLock.lock();
- try {
- pathToInstanceCache.remove(path);
- nodeToInstanceCache.remove(host);
- } finally {
- instanceCacheLock.unlock();
- }
- LOG.debug("Removed path={}, host={} from cache."
- + " pathToInstanceCache:size={}, nodeToInstanceCache:size={}",
- path, host, pathToInstanceCache.size(), nodeToInstanceCache.size());
- }
+ // TODO: this class is completely unnecessary... 1-on-1 mapping with parent.
+ // Remains here as the legacy of the original higher-level interface (getInstance).
+ private static class DynamicServiceInstanceSet implements LlapServiceInstanceSet {
+ private final PathChildrenCache instancesCache;
+ private final LlapZookeeperRegistryImpl parent;
+ private final ServiceRecordMarshal encoder;
- private void putInCache(String key, Map<String, Set<ServiceInstance>> cache,
- ServiceInstance instance) {
- Set<ServiceInstance> instanceSet = cache.get(key);
- if (instanceSet == null) {
- instanceSet = Sets.newHashSet();
- cache.put(key, instanceSet);
+ public DynamicServiceInstanceSet(PathChildrenCache cache,
+ LlapZookeeperRegistryImpl parent, ServiceRecordMarshal encoder) {
+ this.instancesCache = cache;
+ this.parent = parent;
+ this.encoder = encoder;
+ parent.populateCache(instancesCache);
}
- instanceSet.add(instance);
- }
- private class DynamicServiceInstanceSet implements ServiceInstanceSet {
- private final PathChildrenCache instancesCache;
+ @Override
+ public Collection<LlapServiceInstance> getAll() {
+ return parent.getAll();
+ }
- public DynamicServiceInstanceSet(final PathChildrenCache cache) {
- this.instancesCache = cache;
- populateCache();
+ @Override
+ public Collection<LlapServiceInstance> getAllInstancesOrdered(boolean consistentIndexes) {
+ return parent.getAllInstancesOrdered(consistentIndexes, instancesCache);
}
- private void populateCache() {
- for (ChildData childData : instancesCache.getCurrentData()) {
- byte[] data = getWorkerData(childData);
- if (data == null) continue;
- try {
- ServiceRecord srv = encoder.fromBytes(childData.getPath(), data);
- ServiceInstance instance = new DynamicServiceInstance(srv);
- addToCache(childData.getPath(), instance.getHost(), instance);
- } catch (IOException e) {
- LOG.error("Unable to decode data for zkpath: {}." +
- " Ignoring from current instances list..", childData.getPath());
+ @Override
+ public LlapServiceInstance getInstance(String name) {
+ Collection<LlapServiceInstance> instances = getAll();
+ for(LlapServiceInstance instance : instances) {
+ if (instance.getWorkerIdentity().equals(name)) {
+ return instance;
}
}
+ return null;
}
@Override
- public Collection<ServiceInstance> getAll() {
- Set<ServiceInstance> instances = new HashSet<>();
- for(Set<ServiceInstance> instanceSet : pathToInstanceCache.values()) {
- instances.addAll(instanceSet);
- }
- return instances;
+ public Set<LlapServiceInstance> getByHost(String host) {
+ return parent.getByHost(host);
+ }
+
+ @Override
+ public int size() {
+ return parent.size();
}
+ @Override
public ApplicationId getApplicationId() {
for (ChildData childData : instancesCache.getCurrentData()) {
- byte[] data = getWorkerData(childData);
+ byte[] data = getWorkerData(childData, WORKER_PREFIX);
if (data == null) continue;
ServiceRecord sr = null;
try {
@@ -622,147 +308,70 @@ public class LlapZookeeperRegistryImpl implements ServiceRegistry {
}
return null;
}
+ }
- private byte[] getWorkerData(ChildData childData) {
- if (childData == null) return null;
- byte[] data = childData.getData();
- if (data == null) return null;
- if (!extractNodeName(childData).startsWith(WORKER_PREFIX)) return null;
- return data;
- }
-
- @Override
- public Collection<ServiceInstance> getAllInstancesOrdered(boolean consistentIndexes) {
- Map<String, Long> slotByWorker = new HashMap<String, Long>();
- Set<ServiceInstance> unsorted = Sets.newHashSet();
- for (ChildData childData : instancesCache.getCurrentData()) {
- if (childData == null) continue;
- byte[] data = childData.getData();
- if (data == null) continue;
- String nodeName = extractNodeName(childData);
- if (nodeName.startsWith(WORKER_PREFIX)) {
- Set<ServiceInstance> instances = pathToInstanceCache.get(childData.getPath());
- if (instances != null) {
- unsorted.addAll(instances);
- }
- } else if (nodeName.startsWith(SLOT_PREFIX)) {
- slotByWorker.put(extractWorkerIdFromSlot(childData),
- Long.parseLong(nodeName.substring(SLOT_PREFIX.length())));
- } else {
- LOG.info("Ignoring unknown node {}", childData.getPath());
- }
- }
-
- TreeMap<Long, ServiceInstance> sorted = new TreeMap<>();
- long maxSlot = Long.MIN_VALUE;
- for (ServiceInstance worker : unsorted) {
- Long slot = slotByWorker.get(worker.getWorkerIdentity());
- if (slot == null) {
- LOG.info("Unknown slot for {}", worker.getWorkerIdentity());
- continue;
- }
- maxSlot = Math.max(maxSlot, slot);
- sorted.put(slot, worker);
- }
+ private static String extractWorkerIdFromSlot(ChildData childData) {
+ return new String(childData.getData(), SlotZnode.CHARSET);
+ }
- if (consistentIndexes) {
- // Add dummy instances to all slots where LLAPs are MIA... I can haz insert_iterator?
- TreeMap<Long, ServiceInstance> dummies = new TreeMap<>();
- Iterator<Long> keyIter = sorted.keySet().iterator();
- long expected = 0;
- Long ts = null;
- while (keyIter.hasNext()) {
- Long slot = keyIter.next();
- assert slot >= expected;
- while (slot > expected) {
- if (ts == null) {
- ts = System.nanoTime(); // Inactive nodes restart every call!
- }
- dummies.put(expected, new InactiveServiceInstance("inactive-" + expected + "-" + ts));
- ++expected;
- }
- ++expected;
+ // The real implementation for the instanceset... instanceset has its own copy of the
+ // ZK cache yet completely depends on the parent in every other aspect and is thus unneeded.
+
+ Collection<LlapServiceInstance> getAllInstancesOrdered(
+ boolean consistentIndexes, PathChildrenCache instancesCache) {
+ Map<String, Long> slotByWorker = new HashMap<String, Long>();
+ Set<LlapServiceInstance> unsorted = Sets.newHashSet();
+ for (ChildData childData : instancesCache.getCurrentData()) {
+ if (childData == null) continue;
+ byte[] data = childData.getData();
+ if (data == null) continue;
+ String nodeName = extractNodeName(childData);
+ if (nodeName.startsWith(WORKER_PREFIX)) {
+ Set<LlapServiceInstance> instances = getInstancesByPath(childData.getPath());
+ if (instances != null) {
+ unsorted.addAll(instances);
}
- sorted.putAll(dummies);
+ } else if (nodeName.startsWith(SLOT_PREFIX)) {
+ slotByWorker.put(extractWorkerIdFromSlot(childData),
+ Long.parseLong(nodeName.substring(SLOT_PREFIX.length())));
+ } else {
+ LOG.info("Ignoring unknown node {}", childData.getPath());
}
- return sorted.values();
}
- @Override
- public ServiceInstance getInstance(String name) {
- Collection<ServiceInstance> instances = getAll();
- for(ServiceInstance instance : instances) {
- if (instance.getWorkerIdentity().equals(name)) {
- return instance;
- }
+ TreeMap<Long, LlapServiceInstance> sorted = new TreeMap<>();
+ long maxSlot = Long.MIN_VALUE;
+ for (LlapServiceInstance worker : unsorted) {
+ Long slot = slotByWorker.get(worker.getWorkerIdentity());
+ if (slot == null) {
+ LOG.info("Unknown slot for {}", worker.getWorkerIdentity());
+ continue;
}
- return null;
- }
-
- @Override
- public Set<ServiceInstance> getByHost(String host) {
- Set<ServiceInstance> byHost = nodeToInstanceCache.get(host);
- byHost = (byHost == null) ? Sets.<ServiceInstance>newHashSet() : byHost;
- if (LOG.isDebugEnabled()) {
- LOG.debug("Returning " + byHost.size() + " hosts for locality allocation on " + host);
- }
- return byHost;
- }
-
- @Override
- public int size() {
- // not using the path child cache here as there could be more than 1 path per host (worker and slot znodes)
- return nodeToInstanceCache.size();
- }
- }
-
- // TODO: make class static? fields leak
- private class InstanceStateChangeListener implements PathChildrenCacheListener {
- private final Logger LOG = LoggerFactory.getLogger(InstanceStateChangeListener.class);
-
- @Override
- public void childEvent(final CuratorFramework client, final PathChildrenCacheEvent event)
- throws Exception {
- Preconditions.checkArgument(client != null
- && client.getState() == CuratorFrameworkState.STARTED, "client is not started");
-
- synchronized (this) {
- ChildData childData = event.getData();
- if (childData == null)
- return;
- String nodeName = extractNodeName(childData);
- if (!nodeName.startsWith(WORKER_PREFIX))
- return; // No need to propagate slot updates.
- LOG.info("{} for zknode {} in llap namespace", event.getType(), childData.getPath());
- ServiceInstance instance = extractServiceInstance(event, childData);
- switch (event.getType()) {
- case CHILD_ADDED:
- addToCache(childData.getPath(), instance.getHost(), instance);
- for (ServiceInstanceStateChangeListener listener : stateChangeListeners) {
- listener.onCreate(instance);
- }
- break;
- case CHILD_UPDATED:
- addToCache(childData.getPath(), instance.getHost(), instance);
- for (ServiceInstanceStateChangeListener listener : stateChangeListeners) {
- listener.onUpdate(instance);
- }
- break;
- case CHILD_REMOVED:
- removeFromCache(childData.getPath(), instance.getHost());
- for (ServiceInstanceStateChangeListener listener : stateChangeListeners) {
- listener.onRemove(instance);
+ maxSlot = Math.max(maxSlot, slot);
+ sorted.put(slot, worker);
+ }
+
+ if (consistentIndexes) {
+ // Add dummy instances to all slots where LLAPs are MIA... I can haz insert_iterator?
+ TreeMap<Long, LlapServiceInstance> dummies = new TreeMap<>();
+ Iterator<Long> keyIter = sorted.keySet().iterator();
+ long expected = 0;
+ Long ts = null;
+ while (keyIter.hasNext()) {
+ Long slot = keyIter.next();
+ assert slot >= expected;
+ while (slot > expected) {
+ if (ts == null) {
+ ts = System.nanoTime(); // Inactive nodes restart every call!
}
- break;
- default:
- // Ignore all the other events; logged above.
+ dummies.put(expected, new InactiveServiceInstance("inactive-" + expected + "-" + ts));
+ ++expected;
}
+ ++expected;
}
+ sorted.putAll(dummies);
}
- }
-
- private static String extractWorkerIdFromSlot(ChildData childData) {
- return new String(childData.getData(), SlotZnode.CHARSET);
+ return sorted.values();
}
private static String extractNodeName(ChildData childData) {
@@ -774,187 +383,44 @@ public class LlapZookeeperRegistryImpl implements ServiceRegistry {
return nodeName;
}
- private ServiceInstance extractServiceInstance(
- PathChildrenCacheEvent event, ChildData childData) {
- byte[] data = childData.getData();
- if (data == null) return null;
- try {
- ServiceRecord srv = encoder.fromBytes(event.getData().getPath(), data);
- return new DynamicServiceInstance(srv);
- } catch (IOException e) {
- LOG.error("Unable to decode data for zknode: {}." +
- " Dropping notification of type: {}", childData.getPath(), event.getType());
- return null;
- }
- }
-
+
@Override
- public ServiceInstanceSet getInstances(
+ public LlapServiceInstanceSet getInstances(
String component, long clusterReadyTimeoutMs) throws IOException {
- checkPathChildrenCache(clusterReadyTimeoutMs);
+ PathChildrenCache instancesCache = ensureInstancesCache(clusterReadyTimeoutMs);
// lazily create instances
if (instances == null) {
- this.instances = new DynamicServiceInstanceSet(instancesCache);
+ this.instances = new DynamicServiceInstanceSet(instancesCache, this, encoder);
}
return instances;
}
@Override
public ApplicationId getApplicationId() throws IOException {
- getInstances("LLAP", 0);
- return instances.getApplicationId();
- }
-
- @Override
- public synchronized void registerStateChangeListener(
- final ServiceInstanceStateChangeListener listener)
- throws IOException {
- checkPathChildrenCache(0);
-
- this.stateChangeListeners.add(listener);
- }
-
- private synchronized void checkPathChildrenCache(long clusterReadyTimeoutMs) throws IOException {
- Preconditions.checkArgument(zooKeeperClient != null &&
- zooKeeperClient.getState() == CuratorFrameworkState.STARTED, "client is not started");
- // lazily create PathChildrenCache
- if (instancesCache != null) return;
- ExecutorService tp = Executors.newFixedThreadPool(1, new ThreadFactoryBuilder()
- .setDaemon(true).setNameFormat("StateChangeNotificationHandler").build());
- long startTimeNs = System.nanoTime(), deltaNs = clusterReadyTimeoutMs * 1000000L;
- long sleepTimeMs = Math.min(16, clusterReadyTimeoutMs);
- while (true) {
- PathChildrenCache instancesCache = new PathChildrenCache(zooKeeperClient, workersPath, true);
- instancesCache.getListenable().addListener(new InstanceStateChangeListener(), tp);
- try {
- instancesCache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);
- this.instancesCache = instancesCache;
- break;
- } catch (InvalidACLException e) {
- // PathChildrenCache tried to mkdir when the znode wasn't there, and failed.
- CloseableUtils.closeQuietly(instancesCache);
- long elapsedNs = System.nanoTime() - startTimeNs;
- if (deltaNs == 0 || deltaNs <= elapsedNs) {
- LOG.error("Unable to start curator PathChildrenCache", e);
- throw new IOException(e);
- }
- LOG.warn("The cluster is not started yet (InvalidACL); will retry");
- try {
- Thread.sleep(Math.min(sleepTimeMs, (deltaNs - elapsedNs)/1000000L));
- } catch (InterruptedException e1) {
- LOG.error("Interrupted while retrying the PathChildrenCache startup");
- throw new IOException(e1);
- }
- sleepTimeMs = sleepTimeMs << 1;
- } catch (Exception e) {
- CloseableUtils.closeQuietly(instancesCache);
- LOG.error("Unable to start curator PathChildrenCache", e);
- throw new IOException(e);
- }
- }
+ return getInstances("LLAP", 0).getApplicationId();
}
@Override
public void start() throws IOException {
- if (zooKeeperClient != null) {
- setupZookeeperAuth(this.conf);
- zooKeeperClient.start();
- }
- // Init closeable utils in case register is not called (see HIVE-13322)
- CloseableUtils.class.getName();
+ super.start();
}
@Override
public void stop() throws IOException {
- CloseableUtils.closeQuietly(znode);
+ super.stop();
CloseableUtils.closeQuietly(slotZnode);
- CloseableUtils.closeQuietly(instancesCache);
- CloseableUtils.closeQuietly(zooKeeperClient);
}
-
- private void setupZookeeperAuth(final Configuration conf) throws IOException {
- if (UserGroupInformation.isSecurityEnabled() && LlapProxy.isDaemon()) {
- LOG.info("UGI security is enabled. Setting up ZK auth.");
-
- String llapPrincipal = HiveConf.getVar(conf, ConfVars.LLAP_KERBEROS_PRINCIPAL);
- if (llapPrincipal == null || llapPrincipal.isEmpty()) {
- throw new IOException("Llap Kerberos principal is empty");
- }
-
- String llapKeytab = HiveConf.getVar(conf, ConfVars.LLAP_KERBEROS_KEYTAB_FILE);
- if (llapKeytab == null || llapKeytab.isEmpty()) {
- throw new IOException("Llap Kerberos keytab is empty");
- }
-
- // Install the JAAS Configuration for the runtime
- setZookeeperClientKerberosJaasConfig(llapPrincipal, llapKeytab);
- } else {
- LOG.info("UGI security is not enabled, or non-daemon environment. Skipping setting up ZK auth.");
- }
- }
-
- /**
- * Dynamically sets up the JAAS configuration that uses kerberos
- *
- * @param principal
- * @param keyTabFile
- * @throws IOException
- */
- private void setZookeeperClientKerberosJaasConfig(String principal, String keyTabFile)
- throws IOException {
- // ZooKeeper property name to pick the correct JAAS conf section
- final String SASL_LOGIN_CONTEXT_NAME = "LlapZooKeeperClient";
- System.setProperty(ZooKeeperSaslClient.LOGIN_CONTEXT_NAME_KEY, SASL_LOGIN_CONTEXT_NAME);
-
- principal = SecurityUtil.getServerPrincipal(principal, "0.0.0.0");
- userNameFromPrincipal = LlapUtil.getUserNameFromPrincipal(principal);
- JaasConfiguration jaasConf = new JaasConfiguration(SASL_LOGIN_CONTEXT_NAME, principal,
- keyTabFile);
-
- // Install the Configuration in the runtime.
- javax.security.auth.login.Configuration.setConfiguration(jaasConf);
+ @Override
+ protected LlapServiceInstance createServiceInstance(ServiceRecord srv) throws IOException {
+ return new DynamicServiceInstance(srv);
}
- /**
- * A JAAS configuration for ZooKeeper clients intended to use for SASL
- * Kerberos.
- */
- private static class JaasConfiguration extends javax.security.auth.login.Configuration {
- // Current installed Configuration
- private final javax.security.auth.login.Configuration baseConfig = javax.security.auth.login.Configuration
- .getConfiguration();
- private final String loginContextName;
- private final String principal;
- private final String keyTabFile;
-
- public JaasConfiguration(String llapLoginContextName, String principal, String keyTabFile) {
- this.loginContextName = llapLoginContextName;
- this.principal = principal;
- this.keyTabFile = keyTabFile;
- }
-
- @Override
- public AppConfigurationEntry[] getAppConfigurationEntry(String appName) {
- if (loginContextName.equals(appName)) {
- Map<String, String> krbOptions = new HashMap<String, String>();
- krbOptions.put("doNotPrompt", "true");
- krbOptions.put("storeKey", "true");
- krbOptions.put("useKeyTab", "true");
- krbOptions.put("principal", principal);
- krbOptions.put("keyTab", keyTabFile);
- krbOptions.put("refreshKrb5Config", "true");
- AppConfigurationEntry llapZooKeeperClientEntry = new AppConfigurationEntry(
- KerberosUtil.getKrb5LoginModuleName(),
- AppConfigurationEntry.LoginModuleControlFlag.REQUIRED, krbOptions);
- return new AppConfigurationEntry[]{llapZooKeeperClientEntry};
- }
- // Try the base config
- if (baseConfig != null) {
- return baseConfig.getAppConfigurationEntry(appName);
- }
- return null;
- }
+ @Override
+ protected String getZkPathUser(Configuration conf) {
+ // External LLAP clients would need to set LLAP_ZK_REGISTRY_USER to the LLAP daemon user (hive),
+ // rather than relying on RegistryUtils.currentUser().
+ return HiveConf.getVar(conf, ConfVars.LLAP_ZK_REGISTRY_USER, RegistryUtils.currentUser());
}
}
http://git-wip-us.apache.org/repos/asf/hive/blob/4795b996/llap-client/src/java/org/apache/hadoop/hive/llap/security/LlapTokenClient.java
----------------------------------------------------------------------
diff --git a/llap-client/src/java/org/apache/hadoop/hive/llap/security/LlapTokenClient.java b/llap-client/src/java/org/apache/hadoop/hive/llap/security/LlapTokenClient.java
index ace9475..783a19f 100644
--- a/llap-client/src/java/org/apache/hadoop/hive/llap/security/LlapTokenClient.java
+++ b/llap-client/src/java/org/apache/hadoop/hive/llap/security/LlapTokenClient.java
@@ -33,8 +33,8 @@ import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.llap.impl.LlapManagementProtocolClientImpl;
import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.GetTokenRequestProto;
import org.apache.hadoop.hive.llap.registry.impl.LlapRegistryService;
-import org.apache.hadoop.hive.llap.registry.ServiceInstance;
-import org.apache.hadoop.hive.llap.registry.ServiceInstanceSet;
+import org.apache.hadoop.hive.llap.registry.LlapServiceInstance;
+import org.apache.hadoop.hive.llap.registry.LlapServiceInstanceSet;
import org.apache.hadoop.hive.llap.security.LlapTokenIdentifier;
import org.apache.hadoop.io.DataInputByteBuffer;
import org.apache.hadoop.io.retry.RetryPolicies;
@@ -55,10 +55,10 @@ public class LlapTokenClient {
private final SocketFactory socketFactory;
private final RetryPolicy retryPolicy;
private final Configuration conf;
- private ServiceInstanceSet activeInstances;
- private Collection<ServiceInstance> lastKnownInstances;
+ private LlapServiceInstanceSet activeInstances;
+ private Collection<LlapServiceInstance> lastKnownInstances;
private LlapManagementProtocolClientImpl client;
- private ServiceInstance clientInstance;
+ private LlapServiceInstance clientInstance;
public LlapTokenClient(Configuration conf) {
this.conf = conf;
@@ -71,7 +71,7 @@ public class LlapTokenClient {
public Token<LlapTokenIdentifier> getDelegationToken(String appId) throws IOException {
if (!UserGroupInformation.isSecurityEnabled()) return null;
- Iterator<ServiceInstance> llaps = null;
+ Iterator<LlapServiceInstance> llaps = null;
if (clientInstance == null) {
assert client == null;
llaps = getLlapServices(false).iterator();
@@ -128,7 +128,7 @@ public class LlapTokenClient {
}
/** Synchronized - LLAP registry and instance set are not thread safe. */
- private synchronized List<ServiceInstance> getLlapServices(
+ private synchronized List<LlapServiceInstance> getLlapServices(
boolean doForceRefresh) throws IOException {
if (!doForceRefresh && lastKnownInstances != null) {
return new ArrayList<>(lastKnownInstances);
@@ -137,12 +137,12 @@ public class LlapTokenClient {
registry.start();
activeInstances = registry.getInstances();
}
- Collection<ServiceInstance> daemons = activeInstances.getAll();
+ Collection<LlapServiceInstance> daemons = activeInstances.getAll();
if (daemons == null || daemons.isEmpty()) {
throw new RuntimeException("No LLAPs found");
}
lastKnownInstances = daemons;
- return new ArrayList<ServiceInstance>(lastKnownInstances);
+ return new ArrayList<LlapServiceInstance>(lastKnownInstances);
}
}
http://git-wip-us.apache.org/repos/asf/hive/blob/4795b996/llap-client/src/java/org/apache/hadoop/hive/registry/ServiceInstance.java
----------------------------------------------------------------------
diff --git a/llap-client/src/java/org/apache/hadoop/hive/registry/ServiceInstance.java b/llap-client/src/java/org/apache/hadoop/hive/registry/ServiceInstance.java
new file mode 100644
index 0000000..908b3bb
--- /dev/null
+++ b/llap-client/src/java/org/apache/hadoop/hive/registry/ServiceInstance.java
@@ -0,0 +1,47 @@
+/*
+ * Licensed 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.registry;
+
+import java.util.Map;
+
+public interface ServiceInstance {
+
+ /**
+ * Worker identity is a UUID (unique across restarts), to identify a node which died & was brought
+ * back on the same host/port
+ */
+ public abstract String getWorkerIdentity();
+
+ /**
+ * Hostname of the service instance
+ *
+ * @return
+ */
+ public abstract String getHost();
+
+ /**
+ * RPC Endpoint for service instance
+ *
+ * @return
+ */
+ public int getRpcPort();
+
+ /**
+ * Config properties of the Service Instance (llap.daemon.*)
+ *
+ * @return
+ */
+ public abstract Map<String, String> getProperties();
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hive/blob/4795b996/llap-client/src/java/org/apache/hadoop/hive/registry/ServiceInstanceSet.java
----------------------------------------------------------------------
diff --git a/llap-client/src/java/org/apache/hadoop/hive/registry/ServiceInstanceSet.java b/llap-client/src/java/org/apache/hadoop/hive/registry/ServiceInstanceSet.java
new file mode 100644
index 0000000..34fba5c
--- /dev/null
+++ b/llap-client/src/java/org/apache/hadoop/hive/registry/ServiceInstanceSet.java
@@ -0,0 +1,61 @@
+/*
+ * Licensed 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.registry;
+
+import java.util.Collection;
+import java.util.Set;
+
+/**
+ * Note: For most of the implementations, there's no guarantee that the ServiceInstance returned by
+ * one invocation is the same as the instance returned by another invocation. e.g. the ZK registry
+ * returns a new ServiceInstance object each time a getInstance call is made.
+ */
+public interface ServiceInstanceSet<InstanceType extends ServiceInstance> {
+
+ /**
+ * Get an instance mapping which map worker identity to each instance.
+ *
+ * The worker identity does not collide between restarts, so each restart will have a unique id,
+ * while having the same host/ip pair.
+ *
+ * @return
+ */
+ Collection<InstanceType> getAll();
+
+ /**
+ * Get an instance by worker identity.
+ *
+ * @param name
+ * @return
+ */
+ InstanceType getInstance(String name);
+
+ /**
+ * Get a list of service instances for a given host.
+ *
+ * The list could include dead and alive instances.
+ *
+ * @param host
+ * @return
+ */
+ Set<InstanceType> getByHost(String host);
+
+ /**
+ * Get number of instances in the currently availabe.
+ *
+ * @return - number of instances
+ */
+ int size();
+
+}
http://git-wip-us.apache.org/repos/asf/hive/blob/4795b996/llap-client/src/java/org/apache/hadoop/hive/registry/ServiceInstanceStateChangeListener.java
----------------------------------------------------------------------
diff --git a/llap-client/src/java/org/apache/hadoop/hive/registry/ServiceInstanceStateChangeListener.java b/llap-client/src/java/org/apache/hadoop/hive/registry/ServiceInstanceStateChangeListener.java
new file mode 100644
index 0000000..0a44179
--- /dev/null
+++ b/llap-client/src/java/org/apache/hadoop/hive/registry/ServiceInstanceStateChangeListener.java
@@ -0,0 +1,42 @@
+/**
+ * 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.registry;
+
+/**
+ * Callback listener for instance state change events
+ */
+public interface ServiceInstanceStateChangeListener<InstanceType extends ServiceInstance> {
+ /**
+ * Called when new {@link ServiceInstance} is created.
+ *
+ * @param serviceInstance - created service instance
+ */
+ void onCreate(InstanceType serviceInstance);
+
+ /**
+ * Called when an existing {@link ServiceInstance} is updated.
+ *
+ * @param serviceInstance - updated service instance
+ */
+ void onUpdate(InstanceType serviceInstance);
+
+ /**
+ * Called when an existing {@link ServiceInstance} is removed.
+ *
+ * @param serviceInstance - removed service instance
+ */
+ void onRemove(InstanceType serviceInstance);
+}
http://git-wip-us.apache.org/repos/asf/hive/blob/4795b996/llap-client/src/java/org/apache/hadoop/hive/registry/impl/ServiceInstanceBase.java
----------------------------------------------------------------------
diff --git a/llap-client/src/java/org/apache/hadoop/hive/registry/impl/ServiceInstanceBase.java b/llap-client/src/java/org/apache/hadoop/hive/registry/impl/ServiceInstanceBase.java
new file mode 100644
index 0000000..db3d788
--- /dev/null
+++ b/llap-client/src/java/org/apache/hadoop/hive/registry/impl/ServiceInstanceBase.java
@@ -0,0 +1,93 @@
+/*
+ * Licensed 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.registry.impl;
+
+import java.io.IOException;
+import java.util.Map;
+import org.apache.hadoop.hive.registry.ServiceInstance;
+import org.apache.hadoop.registry.client.binding.RegistryTypeUtils;
+import org.apache.hadoop.registry.client.types.AddressTypes;
+import org.apache.hadoop.registry.client.types.Endpoint;
+import org.apache.hadoop.registry.client.types.ServiceRecord;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ServiceInstanceBase implements ServiceInstance {
+ private static final Logger LOG = LoggerFactory.getLogger(ServiceInstanceBase.class);
+
+ protected final ServiceRecord srv;
+ protected final String host;
+ protected final int rpcPort;
+
+ public ServiceInstanceBase(ServiceRecord srv, String rpcName) throws IOException {
+ this.srv = srv;
+
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Working with ServiceRecord: {}", srv);
+ }
+
+ final Endpoint rpc = srv.getInternalEndpoint(rpcName);
+
+ this.host =
+ RegistryTypeUtils.getAddressField(rpc.addresses.get(0),
+ AddressTypes.ADDRESS_HOSTNAME_FIELD);
+ this.rpcPort =
+ Integer.parseInt(RegistryTypeUtils.getAddressField(rpc.addresses.get(0),
+ AddressTypes.ADDRESS_PORT_FIELD));
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+
+ ServiceInstanceBase other = (ServiceInstanceBase) o;
+ return this.getWorkerIdentity().equals(other.getWorkerIdentity());
+ }
+
+ @Override
+ public int hashCode() {
+ return getWorkerIdentity().hashCode();
+ }
+
+ @Override
+ public String getWorkerIdentity() {
+ return srv.get(ZkRegistryBase.UNIQUE_IDENTIFIER);
+ }
+
+ @Override
+ public String getHost() {
+ return host;
+ }
+
+ @Override
+ public int getRpcPort() {
+ return rpcPort;
+ }
+
+ @Override
+ public Map<String, String> getProperties() {
+ return srv.attributes();
+ }
+
+ @Override
+ public String toString() {
+ return "DynamicServiceInstance [id=" + getWorkerIdentity() + ", host="
+ + host + ":" + rpcPort + "]";
+ }
+}
\ No newline at end of file