hive-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ser...@apache.org
Subject [2/2] hive git commit: HIVE-17409 : refactor LLAP ZK registry to make the ZK-registry part reusable (Sergey Shelukhin, reviewed by Prasanth Jayachandran)
Date Fri, 01 Sep 2017 18:48:04 GMT
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 <sershe@apache.org>
Authored: Fri Sep 1 11:44:11 2017 -0700
Committer: sergey <sershe@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 &amp; 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 &amp; 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


Mime
View raw message