hive-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ss...@apache.org
Subject hive git commit: HIVE-12256. Move LLAP registry into llap-client module. (Siddharth Seth, reviewed by Gopal V)
Date Fri, 30 Oct 2015 04:57:36 GMT
Repository: hive
Updated Branches:
  refs/heads/master 8f881f87b -> 215c5770c


HIVE-12256. Move LLAP registry into llap-client module. (Siddharth Seth, reviewed by Gopal V)


Project: http://git-wip-us.apache.org/repos/asf/hive/repo
Commit: http://git-wip-us.apache.org/repos/asf/hive/commit/215c5770
Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/215c5770
Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/215c5770

Branch: refs/heads/master
Commit: 215c5770c7ef5e0d1428f9288354205eedcee399
Parents: 8f881f8
Author: Siddharth Seth <sseth@apache.org>
Authored: Thu Oct 29 21:56:38 2015 -0700
Committer: Siddharth Seth <sseth@apache.org>
Committed: Thu Oct 29 21:56:38 2015 -0700

----------------------------------------------------------------------
 llap-client/pom.xml                             |   6 +
 .../hive/llap/registry/ServiceInstance.java     |  73 ++++
 .../hive/llap/registry/ServiceInstanceSet.java  |  57 +++
 .../hive/llap/registry/ServiceRegistry.java     |  59 +++
 .../registry/impl/LlapFixedRegistryImpl.java    | 223 +++++++++++
 .../llap/registry/impl/LlapRegistryService.java |  87 +++++
 .../registry/impl/LlapYarnRegistryImpl.java     | 383 ++++++++++++++++++
 llap-server/pom.xml                             |   6 -
 .../hive/llap/daemon/impl/LlapDaemon.java       |   2 +-
 .../llap/daemon/registry/ServiceInstance.java   |  73 ----
 .../daemon/registry/ServiceInstanceSet.java     |  57 ---
 .../llap/daemon/registry/ServiceRegistry.java   |  59 ---
 .../registry/impl/LlapFixedRegistryImpl.java    | 223 -----------
 .../registry/impl/LlapRegistryService.java      |  87 -----
 .../registry/impl/LlapYarnRegistryImpl.java     | 384 -------------------
 .../dag/app/rm/LlapTaskSchedulerService.java    |   6 +-
 .../app/rm/TestLlapTaskSchedulerService.java    |   2 +-
 17 files changed, 893 insertions(+), 894 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hive/blob/215c5770/llap-client/pom.xml
----------------------------------------------------------------------
diff --git a/llap-client/pom.xml b/llap-client/pom.xml
index 02243f8..0cd2ec9 100644
--- a/llap-client/pom.xml
+++ b/llap-client/pom.xml
@@ -72,6 +72,12 @@
       <version>${hadoop.version}</version>
       <optional>true</optional>
     </dependency>
+    <dependency>
+      <groupId>org.apache.hadoop</groupId>
+      <artifactId>hadoop-yarn-registry</artifactId>
+      <version>${hadoop.version}</version>
+      <optional>true</optional>
+    </dependency>
     <!-- test inter-project -->
     <dependency>
       <groupId>junit</groupId>

http://git-wip-us.apache.org/repos/asf/hive/blob/215c5770/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
new file mode 100644
index 0000000..f116de4
--- /dev/null
+++ b/llap-client/src/java/org/apache/hadoop/hive/llap/registry/ServiceInstance.java
@@ -0,0 +1,73 @@
+/*
+ * 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();
+
+  /**
+   * Shuffle Endpoint for service instance
+   * 
+   * @return
+   */
+  public int getShufflePort();
+
+  /**
+   * Return the last known state (without refreshing)
+   * 
+   * @return
+   */
+
+  public boolean isAlive();
+
+  /**
+   * 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/215c5770/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
new file mode 100644
index 0000000..388b5f3
--- /dev/null
+++ b/llap-client/src/java/org/apache/hadoop/hive/llap/registry/ServiceInstanceSet.java
@@ -0,0 +1,57 @@
+/*
+ * 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.io.IOException;
+import java.util.Map;
+import java.util.Set;
+
+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
+   */
+  public Map<String, ServiceInstance> getAll();
+
+  /**
+   * Get an instance by worker identity.
+   * 
+   * @param name
+   * @return
+   */
+  public 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
+   */
+  public Set<ServiceInstance> getByHost(String host);
+
+  /**
+   * Refresh the instance set from registry backing store.
+   * 
+   * @throws IOException
+   */
+  public void refresh() throws IOException;
+
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/hive/blob/215c5770/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
new file mode 100644
index 0000000..d3fb517
--- /dev/null
+++ b/llap-client/src/java/org/apache/hadoop/hive/llap/registry/ServiceRegistry.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 java.io.IOException;
+
+/**
+ * ServiceRegistry interface for switching between fixed host and dynamic registry implementations.
+ */
+public interface ServiceRegistry {
+
+  /**
+   * Start the service registry
+   * 
+   * @throws InterruptedException
+   */
+  public void start() throws InterruptedException;
+
+  /**
+   * Stop the service registry
+   * 
+   * @throws InterruptedException
+   */
+  public void stop() throws InterruptedException;
+
+  /**
+   * Register the current instance - the implementation takes care of the endpoints to register.
+   * 
+   * @throws IOException
+   */
+  public void register() throws IOException;
+
+  /**
+   * Remove the current registration cleanly (implementation defined cleanup)
+   * 
+   * @throws IOException
+   */
+  public void unregister() throws IOException;
+
+  /**
+   * Client API to get the list of instances registered via the current registry key.
+   * 
+   * @param component
+   * @return
+   * @throws IOException
+   */
+  public ServiceInstanceSet getInstances(String component) throws IOException;
+}

http://git-wip-us.apache.org/repos/asf/hive/blob/215c5770/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
new file mode 100644
index 0000000..34e0682
--- /dev/null
+++ b/llap-client/src/java/org/apache/hadoop/hive/llap/registry/impl/LlapFixedRegistryImpl.java
@@ -0,0 +1,223 @@
+/*
+ * 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.impl;
+
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.UnknownHostException;
+import java.util.HashMap;
+import java.util.HashSet;
+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.llap.configuration.LlapConfiguration;
+import org.apache.hadoop.hive.llap.registry.ServiceInstance;
+import org.apache.hadoop.hive.llap.registry.ServiceInstanceSet;
+import org.apache.hadoop.hive.llap.registry.ServiceRegistry;
+import org.apache.hadoop.net.NetUtils;
+import org.apache.hadoop.util.StringUtils;
+import org.apache.hadoop.yarn.api.records.Resource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class LlapFixedRegistryImpl implements ServiceRegistry {
+
+  private static final Logger LOG = LoggerFactory.getLogger(LlapFixedRegistryImpl.class);
+
+  @InterfaceAudience.Private
+  // This is primarily for testing to avoid the host lookup
+  public static final String FIXED_REGISTRY_RESOLVE_HOST_NAMES = "fixed.registry.resolve.host.names";
+
+  private final int port;
+  private final int shuffle;
+  private final String[] hosts;
+  private final int memory;
+  private final int vcores;
+  private final boolean resolveHosts;
+
+  private final Map<String, String> srv = new HashMap<String, String>();
+
+  public LlapFixedRegistryImpl(String hosts, Configuration conf) {
+    this.hosts = hosts.split(",");
+    this.port =
+        conf.getInt(LlapConfiguration.LLAP_DAEMON_RPC_PORT,
+            LlapConfiguration.LLAP_DAEMON_RPC_PORT_DEFAULT);
+    this.shuffle =
+        conf.getInt(LlapConfiguration.LLAP_DAEMON_YARN_SHUFFLE_PORT,
+            LlapConfiguration.LLAP_DAEMON_YARN_SHUFFLE_PORT_DEFAULT);
+    this.resolveHosts = conf.getBoolean(FIXED_REGISTRY_RESOLVE_HOST_NAMES, true);
+
+    for (Map.Entry<String, String> kv : conf) {
+      if (kv.getKey().startsWith(LlapConfiguration.LLAP_DAEMON_PREFIX)
+          || kv.getKey().startsWith("hive.llap.")
+          || kv.getKey().startsWith(LlapConfiguration.LLAP_PREFIX)) {
+        // TODO: read this somewhere useful, like the task scheduler
+        srv.put(kv.getKey(), kv.getValue());
+      }
+    }
+
+    this.memory =
+        conf.getInt(LlapConfiguration.LLAP_DAEMON_MEMORY_PER_INSTANCE_MB,
+            LlapConfiguration.LLAP_DAEMON_MEMORY_PER_INSTANCE_MB_DEFAULT);
+    this.vcores =
+        conf.getInt(LlapConfiguration.LLAP_DAEMON_NUM_EXECUTORS,
+            LlapConfiguration.LLAP_DAEMON_NUM_EXECUTORS_DEFAULT);
+  }
+
+  @Override
+  public void start() throws InterruptedException {
+    // nothing to start
+  }
+
+  @Override
+  public void stop() throws InterruptedException {
+    // nothing to stop
+  }
+
+  @Override
+  public void register() throws IOException {
+    // nothing to register
+  }
+
+  @Override
+  public void unregister() throws IOException {
+    // nothing to unregister
+  }
+
+  public static String getWorkerIdentity(String host) {
+    // trigger clean errors for anyone who mixes up identity with hosts
+    return "host-" + host;
+  }
+
+  private final class FixedServiceInstance implements ServiceInstance {
+
+    private final String host;
+
+    public FixedServiceInstance(String host) {
+      if (resolveHosts) {
+        try {
+          InetAddress inetAddress = InetAddress.getByName(host);
+          if (NetUtils.isLocalAddress(inetAddress)) {
+            InetSocketAddress socketAddress = new InetSocketAddress(0);
+            socketAddress = NetUtils.getConnectAddress(socketAddress);
+            LOG.info("Adding host identified as local: " + host + " as "
+                + socketAddress.getHostName());
+            host = socketAddress.getHostName();
+          }
+        } catch (UnknownHostException e) {
+          LOG.warn("Ignoring resolution issues for host: " + host, e);
+        }
+      }
+      this.host = host;
+    }
+
+    @Override
+    public String getWorkerIdentity() {
+      return LlapFixedRegistryImpl.getWorkerIdentity(host);
+    }
+
+    @Override
+    public String getHost() {
+      return host;
+    }
+
+    @Override
+    public int getRpcPort() {
+      // TODO: allow >1 port per host?
+      return LlapFixedRegistryImpl.this.port;
+    }
+
+    @Override
+    public int getShufflePort() {
+      return LlapFixedRegistryImpl.this.shuffle;
+    }
+
+    @Override
+    public boolean isAlive() {
+      return true;
+    }
+
+    @Override
+    public Map<String, String> getProperties() {
+      Map<String, String> properties = new HashMap<>(srv);
+      // no worker identity
+      return properties;
+    }
+
+    @Override
+    public Resource getResource() {
+      return Resource.newInstance(memory, vcores);
+    }
+
+    @Override
+    public String toString() {
+      return "FixedServiceInstance{" +
+          "host=" + host +
+          ", memory=" + memory +
+          ", vcores=" + vcores +
+          '}';
+    }
+  }
+
+  private final class FixedServiceInstanceSet implements ServiceInstanceSet {
+
+    private final Map<String, ServiceInstance> instances = new HashMap<String, ServiceInstance>();
+
+    public FixedServiceInstanceSet() {
+      for (String host : hosts) {
+        // trigger bugs in anyone who uses this as a hostname
+        instances.put(getWorkerIdentity(host), new FixedServiceInstance(host));
+      }
+    }
+
+    @Override
+    public Map<String, ServiceInstance> getAll() {
+      return instances;
+    }
+
+    @Override
+    public ServiceInstance 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));
+      if (inst != null) {
+        byHost.add(inst);
+      }
+      return byHost;
+    }
+
+    @Override
+    public void refresh() throws IOException {
+      // I will do no such thing
+    }
+
+  }
+
+  @Override
+  public ServiceInstanceSet getInstances(String component) throws IOException {
+    return new FixedServiceInstanceSet();
+  }
+
+  @Override
+  public String toString() {
+    return String.format("FixedRegistry hosts=%s", StringUtils.join(",", this.hosts));
+  }
+}

http://git-wip-us.apache.org/repos/asf/hive/blob/215c5770/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
new file mode 100644
index 0000000..a8e1465
--- /dev/null
+++ b/llap-client/src/java/org/apache/hadoop/hive/llap/registry/impl/LlapRegistryService.java
@@ -0,0 +1,87 @@
+/*
+ * 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.impl;
+
+import java.io.IOException;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.llap.configuration.LlapConfiguration;
+import org.apache.hadoop.hive.llap.registry.ServiceInstanceSet;
+import org.apache.hadoop.hive.llap.registry.ServiceRegistry;
+import org.apache.hadoop.service.AbstractService;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class LlapRegistryService extends AbstractService {
+
+  private static final Logger LOG = LoggerFactory.getLogger(LlapRegistryService.class);
+
+  private ServiceRegistry registry = null;
+  private final boolean isDaemon;
+
+  public LlapRegistryService(boolean isDaemon) {
+    super("LlapRegistryService");
+    this.isDaemon = isDaemon;
+  }
+
+  @Override
+  public void serviceInit(Configuration conf) {
+    String hosts = conf.getTrimmed(LlapConfiguration.LLAP_DAEMON_SERVICE_HOSTS);
+    if (hosts.startsWith("@")) {
+      registry = new LlapYarnRegistryImpl(hosts.substring(1), conf, isDaemon);
+    } else {
+      registry = new LlapFixedRegistryImpl(hosts, conf);
+    }
+    LOG.info("Using LLAP registry type " + registry);
+  }
+
+
+  @Override
+  public void serviceStart() throws Exception {
+    if (this.registry != null) {
+      this.registry.start();
+    }
+    if (isDaemon) {
+      registerWorker();
+    }
+  }
+
+  @Override
+  public void serviceStop() throws Exception {
+    if (isDaemon) {
+      unregisterWorker();
+    }
+    if (this.registry != null) {
+      this.registry.stop();
+    } else {
+      LOG.warn("Stopping non-existent registry service");
+    }
+  }
+
+  private void registerWorker() throws IOException {
+    if (this.registry != null) {
+      this.registry.register();
+    }
+  }
+
+  private void unregisterWorker() throws IOException {
+    if (this.registry != null) {
+      this.registry.unregister();
+    }
+  }
+
+  public ServiceInstanceSet getInstances() throws IOException {
+    return this.registry.getInstances("LLAP");
+  }
+}

http://git-wip-us.apache.org/repos/asf/hive/blob/215c5770/llap-client/src/java/org/apache/hadoop/hive/llap/registry/impl/LlapYarnRegistryImpl.java
----------------------------------------------------------------------
diff --git a/llap-client/src/java/org/apache/hadoop/hive/llap/registry/impl/LlapYarnRegistryImpl.java b/llap-client/src/java/org/apache/hadoop/hive/llap/registry/impl/LlapYarnRegistryImpl.java
new file mode 100644
index 0000000..d474b6f
--- /dev/null
+++ b/llap-client/src/java/org/apache/hadoop/hive/llap/registry/impl/LlapYarnRegistryImpl.java
@@ -0,0 +1,383 @@
+/*
+ * 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.impl;
+
+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.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.llap.configuration.LlapConfiguration;
+import org.apache.hadoop.hive.llap.registry.ServiceInstance;
+import org.apache.hadoop.hive.llap.registry.ServiceInstanceSet;
+import org.apache.hadoop.hive.llap.registry.ServiceRegistry;
+import org.apache.hadoop.registry.client.api.RegistryOperationsFactory;
+import org.apache.hadoop.registry.client.binding.RegistryPathUtils;
+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.impl.zk.RegistryOperationsService;
+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.yarn.api.records.Resource;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.zookeeper.CreateMode;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.base.Preconditions;
+
+public class LlapYarnRegistryImpl implements ServiceRegistry {
+
+  private static final Logger LOG = LoggerFactory.getLogger(LlapYarnRegistryImpl.class);
+
+  private final RegistryOperationsService client;
+  private final Configuration conf;
+  private final ServiceRecordMarshal encoder;
+  private final String path;
+
+  private final DynamicServiceInstanceSet instances = new DynamicServiceInstanceSet();
+
+  private static final UUID uniq = UUID.randomUUID();
+  private static final String hostname;
+
+  private static final String UNIQUE_IDENTIFIER = "llap.unique.id";
+
+  private final static String SERVICE_CLASS = "org-apache-hive";
+
+  final ScheduledExecutorService refresher = Executors.newScheduledThreadPool(1,
+      new ThreadFactoryBuilder().setDaemon(true).setNameFormat("LlapYarnRegistryRefresher").build());
+  final long refreshDelay;
+  private final boolean isDaemon;
+
+  static {
+    String localhost = "localhost";
+    try {
+      localhost = InetAddress.getLocalHost().getCanonicalHostName();
+    } catch (UnknownHostException uhe) {
+      // ignore
+    }
+    hostname = localhost;
+  }
+
+  public LlapYarnRegistryImpl(String instanceName, Configuration conf, boolean isDaemon) {
+
+    LOG.info("Llap Registry is enabled with registryid: " + instanceName);
+    this.conf = new Configuration(conf);
+    conf.addResource(YarnConfiguration.YARN_SITE_CONFIGURATION_FILE);
+    // registry reference
+    client = (RegistryOperationsService) RegistryOperationsFactory.createInstance(conf);
+    encoder = new RegistryUtils.ServiceRecordMarshal();
+    this.path = RegistryPathUtils.join(RegistryUtils.componentPath(RegistryUtils.currentUser(),
+        SERVICE_CLASS, instanceName, "workers"), "worker-");
+    refreshDelay =
+        conf.getInt(LlapConfiguration.LLAP_DAEMON_SERVICE_REFRESH_INTERVAL,
+            LlapConfiguration.LLAP_DAEMON_SERVICE_REFRESH_INTERVAL_DEFAULT);
+    this.isDaemon = isDaemon;
+    Preconditions.checkArgument(refreshDelay > 0,
+        "Refresh delay for registry has to be positive = %d", refreshDelay);
+  }
+
+  public Endpoint getRpcEndpoint() {
+    final int rpcPort =
+        conf.getInt(LlapConfiguration.LLAP_DAEMON_RPC_PORT,
+            LlapConfiguration.LLAP_DAEMON_RPC_PORT_DEFAULT);
+    return RegistryTypeUtils.ipcEndpoint("llap", new InetSocketAddress(hostname, rpcPort));
+  }
+
+  public Endpoint getShuffleEndpoint() {
+    final int shufflePort =
+        conf.getInt(LlapConfiguration.LLAP_DAEMON_YARN_SHUFFLE_PORT,
+            LlapConfiguration.LLAP_DAEMON_YARN_SHUFFLE_PORT_DEFAULT);
+    // HTTP today, but might not be
+    return RegistryTypeUtils.inetAddrEndpoint("shuffle", ProtocolTypes.PROTOCOL_TCP, hostname,
+        shufflePort);
+  }
+
+  public Endpoint getServicesEndpoint() {
+    final int servicePort =
+        conf.getInt(LlapConfiguration.LLAP_DAEMON_SERVICE_PORT,
+            LlapConfiguration.LLAP_DAEMON_SERVICE_PORT_DEFAULT);
+    final boolean isSSL =
+        conf.getBoolean(LlapConfiguration.LLAP_DAEMON_SERVICE_SSL,
+            LlapConfiguration.LLAP_DAEMON_SERVICE_SSL_DEFAULT);
+    final String scheme = isSSL ? "https" : "http";
+    final URL serviceURL;
+    try {
+      serviceURL = new URL(scheme, hostname, servicePort, "");
+      return RegistryTypeUtils.webEndpoint("services", serviceURL.toURI());
+    } catch (MalformedURLException e) {
+      throw new RuntimeException(e);
+    } catch (URISyntaxException e) {
+      throw new RuntimeException("llap service URI for " + hostname + " is invalid", e);
+    }
+  }
+
+  private final String getPath() {
+    return this.path;
+  }
+
+  @Override
+  public void register() throws IOException {
+    String path = getPath();
+    ServiceRecord srv = new ServiceRecord();
+    srv.addInternalEndpoint(getRpcEndpoint());
+    srv.addInternalEndpoint(getShuffleEndpoint());
+    srv.addExternalEndpoint(getServicesEndpoint());
+
+    for (Map.Entry<String, String> kv : this.conf) {
+      if (kv.getKey().startsWith(LlapConfiguration.LLAP_DAEMON_PREFIX)
+          || kv.getKey().startsWith("hive.llap.")) {
+        // TODO: read this somewhere useful, like the task scheduler
+        srv.set(kv.getKey(), kv.getValue());
+      }
+    }
+
+    // restart sensitive instance id
+    srv.set(UNIQUE_IDENTIFIER, uniq.toString());
+
+    client.mknode(RegistryPathUtils.parentOf(path), true);
+
+    // FIXME: YARN registry needs to expose Ephemeral_Seq nodes & return the paths
+    client.zkCreate(path, CreateMode.EPHEMERAL_SEQUENTIAL, encoder.toBytes(srv),
+        client.getClientAcls());
+  }
+
+  @Override
+  public void unregister() throws IOException {
+   // Nothing for the zkCreate models
+  }
+
+  private class DynamicServiceInstance implements ServiceInstance {
+
+    private final ServiceRecord srv;
+    private boolean alive = true;
+    private final String host;
+    private final int rpcPort;
+    private final int shufflePort;
+
+    public DynamicServiceInstance(ServiceRecord srv) throws IOException {
+      this.srv = srv;
+
+      final Endpoint shuffle = srv.getInternalEndpoint("shuffle");
+      final Endpoint rpc = srv.getInternalEndpoint("llap");
+
+      this.host =
+          RegistryTypeUtils.getAddressField(rpc.addresses.get(0),
+              AddressTypes.ADDRESS_HOSTNAME_FIELD);
+      this.rpcPort =
+          Integer.valueOf(RegistryTypeUtils.getAddressField(rpc.addresses.get(0),
+              AddressTypes.ADDRESS_PORT_FIELD));
+      this.shufflePort =
+          Integer.valueOf(RegistryTypeUtils.getAddressField(shuffle.addresses.get(0),
+              AddressTypes.ADDRESS_PORT_FIELD));
+    }
+
+    @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;
+    }
+
+    @Override
+    public boolean isAlive() {
+      return alive ;
+    }
+
+    public void kill() {
+      // May be possible to generate a notification back to the scheduler from here.
+      LOG.info("Killing service instance: " + this);
+      this.alive = false;
+    }
+
+    @Override
+    public Map<String, String> getProperties() {
+      return srv.attributes();
+    }
+
+    @Override
+    public Resource getResource() {
+      int memory = Integer.valueOf(srv.get(LlapConfiguration.LLAP_DAEMON_MEMORY_PER_INSTANCE_MB));
+      int vCores = Integer.valueOf(srv.get(LlapConfiguration.LLAP_DAEMON_NUM_EXECUTORS));
+      return Resource.newInstance(memory, vCores);
+    }
+
+    @Override
+    public String toString() {
+      return "DynamicServiceInstance [alive=" + alive + ", host=" + host + ":" + rpcPort + " with resources=" + getResource() +"]";
+    }
+
+    // Relying on the identity hashCode and equality, since refreshing instances retains the old copy
+    // of an already known instance.
+  }
+
+  private class DynamicServiceInstanceSet implements ServiceInstanceSet {
+
+    // LinkedHashMap to retain iteration order.
+    private final Map<String, ServiceInstance> instances = new LinkedHashMap<>();
+
+    @Override
+    public synchronized Map<String, ServiceInstance> getAll() {
+      // Return a copy. Instances may be modified during a refresh.
+      return new LinkedHashMap<>(instances);
+    }
+
+    @Override
+    public synchronized ServiceInstance getInstance(String name) {
+      return instances.get(name);
+    }
+
+    @Override
+    public  void refresh() throws IOException {
+      /* call this from wherever */
+      Map<String, ServiceInstance> freshInstances = new HashMap<String, ServiceInstance>();
+
+      String path = getPath();
+      Map<String, ServiceRecord> records =
+          RegistryUtils.listServiceRecords(client, RegistryPathUtils.parentOf(path));
+      // Synchronize after reading the service records from the external service (ZK)
+      synchronized (this) {
+        Set<String> latestKeys = new HashSet<String>();
+        LOG.info("Starting to refresh ServiceInstanceSet " + System.identityHashCode(this));
+        for (ServiceRecord rec : records.values()) {
+          ServiceInstance instance = new DynamicServiceInstance(rec);
+          if (instance != null) {
+            if (instances != null && instances.containsKey(instance.getWorkerIdentity()) == false) {
+              // add a new object
+              freshInstances.put(instance.getWorkerIdentity(), instance);
+              if (LOG.isInfoEnabled()) {
+                LOG.info("Adding new worker " + instance.getWorkerIdentity() + " which mapped to "
+                    + instance);
+              }
+            } else {
+              if (LOG.isDebugEnabled()) {
+                LOG.debug("Retaining running worker " + instance.getWorkerIdentity() +
+                    " which mapped to " + instance);
+              }
+            }
+          }
+          latestKeys.add(instance.getWorkerIdentity());
+        }
+
+        if (instances != null) {
+          // deep-copy before modifying
+          Set<String> oldKeys = new HashSet<>(instances.keySet());
+          if (oldKeys.removeAll(latestKeys)) {
+            // This is all the records which have not checked in, and are effectively dead.
+            for (String k : oldKeys) {
+              // this is so that people can hold onto ServiceInstance references as placeholders for tasks
+              final DynamicServiceInstance dead = (DynamicServiceInstance) instances.get(k);
+              dead.kill();
+              if (LOG.isInfoEnabled()) {
+                LOG.info("Deleting dead worker " + k + " which mapped to " + dead);
+              }
+            }
+          }
+          // oldKeys contains the set of dead instances at this point.
+          this.instances.keySet().removeAll(oldKeys);
+          this.instances.putAll(freshInstances);
+        } else {
+          this.instances.putAll(freshInstances);
+        }
+      }
+    }
+
+    @Override
+    public synchronized Set<ServiceInstance> getByHost(String host) {
+      // TODO Maybe store this as a map which is populated during construction, to avoid walking
+      // the map on each request.
+      Set<ServiceInstance> byHost = new HashSet<ServiceInstance>();
+
+      for (ServiceInstance i : instances.values()) {
+        if (host.equals(i.getHost())) {
+          // all hosts in instances should be alive in this impl
+          byHost.add(i);
+        }
+        if (LOG.isDebugEnabled()) {
+          LOG.debug("Locality comparing " + host + " to " + i.getHost());
+        }
+      }
+      if (LOG.isDebugEnabled()) {
+        LOG.debug("Returning " + byHost.size() + " hosts for locality allocation on " + host);
+      }
+      return byHost;
+    }
+  }
+
+  @Override
+  public ServiceInstanceSet getInstances(String component) throws IOException {
+    Preconditions.checkArgument("LLAP".equals(component)); // right now there is only 1 component 
+    if (this.client != null) {
+      instances.refresh();
+      return instances;
+    } else {
+      Preconditions.checkNotNull(this.client, "Yarn registry client is not intialized");
+      return null;
+    }
+  }
+
+  @Override
+  public void start() {
+    if (client == null) return;
+    client.start();
+    if (isDaemon) return;
+    refresher.scheduleWithFixedDelay(new Runnable() {
+      @Override
+      public void run() {
+        try {
+          instances.refresh();
+        } catch (IOException ioe) {
+          LOG.warn("Could not refresh hosts during scheduled refresh", ioe);
+        }
+      }
+    }, 0, refreshDelay, TimeUnit.SECONDS);
+  }
+
+  @Override
+  public void stop() {
+    if (client != null) {
+      client.stop();
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/hive/blob/215c5770/llap-server/pom.xml
----------------------------------------------------------------------
diff --git a/llap-server/pom.xml b/llap-server/pom.xml
index 4be45a5..a733e2c 100644
--- a/llap-server/pom.xml
+++ b/llap-server/pom.xml
@@ -98,12 +98,6 @@
       <optional>true</optional>
     </dependency>
     <dependency>
-      <groupId>org.apache.hadoop</groupId>
-      <artifactId>hadoop-yarn-registry</artifactId>
-      <version>${hadoop.version}</version>
-      <optional>true</optional>
-    </dependency>
-    <dependency>
       <groupId>org.apache.tez</groupId>
       <artifactId>tez-runtime-internals</artifactId>
       <version>${tez.version}</version>

http://git-wip-us.apache.org/repos/asf/hive/blob/215c5770/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/LlapDaemon.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/LlapDaemon.java b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/LlapDaemon.java
index 6f75001..98b1ccd 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/LlapDaemon.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/LlapDaemon.java
@@ -31,7 +31,7 @@ import org.apache.hadoop.hive.conf.HiveConf;
 import org.apache.hadoop.hive.llap.configuration.LlapConfiguration;
 import org.apache.hadoop.hive.llap.daemon.ContainerRunner;
 import org.apache.hadoop.hive.llap.daemon.QueryFailedHandler;
-import org.apache.hadoop.hive.llap.daemon.registry.impl.LlapRegistryService;
+import org.apache.hadoop.hive.llap.registry.impl.LlapRegistryService;
 import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.QueryCompleteRequestProto;
 import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.SourceStateUpdatedRequestProto;
 import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.SubmitWorkRequestProto;

http://git-wip-us.apache.org/repos/asf/hive/blob/215c5770/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/registry/ServiceInstance.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/registry/ServiceInstance.java b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/registry/ServiceInstance.java
deleted file mode 100644
index f0f22aa..0000000
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/registry/ServiceInstance.java
+++ /dev/null
@@ -1,73 +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.daemon.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();
-
-  /**
-   * Shuffle Endpoint for service instance
-   * 
-   * @return
-   */
-  public int getShufflePort();
-
-  /**
-   * Return the last known state (without refreshing)
-   * 
-   * @return
-   */
-
-  public boolean isAlive();
-
-  /**
-   * 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/215c5770/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/registry/ServiceInstanceSet.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/registry/ServiceInstanceSet.java b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/registry/ServiceInstanceSet.java
deleted file mode 100644
index 7ab36d4..0000000
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/registry/ServiceInstanceSet.java
+++ /dev/null
@@ -1,57 +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.daemon.registry;
-
-import java.io.IOException;
-import java.util.Map;
-import java.util.Set;
-
-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
-   */
-  public Map<String, ServiceInstance> getAll();
-
-  /**
-   * Get an instance by worker identity.
-   * 
-   * @param name
-   * @return
-   */
-  public 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
-   */
-  public Set<ServiceInstance> getByHost(String host);
-
-  /**
-   * Refresh the instance set from registry backing store.
-   * 
-   * @throws IOException
-   */
-  public void refresh() throws IOException;
-
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/hive/blob/215c5770/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/registry/ServiceRegistry.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/registry/ServiceRegistry.java b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/registry/ServiceRegistry.java
deleted file mode 100644
index a0f9aac..0000000
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/registry/ServiceRegistry.java
+++ /dev/null
@@ -1,59 +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.daemon.registry;
-
-import java.io.IOException;
-
-/**
- * ServiceRegistry interface for switching between fixed host and dynamic registry implementations.
- */
-public interface ServiceRegistry {
-
-  /**
-   * Start the service registry
-   * 
-   * @throws InterruptedException
-   */
-  public void start() throws InterruptedException;
-
-  /**
-   * Stop the service registry
-   * 
-   * @throws InterruptedException
-   */
-  public void stop() throws InterruptedException;
-
-  /**
-   * Register the current instance - the implementation takes care of the endpoints to register.
-   * 
-   * @throws IOException
-   */
-  public void register() throws IOException;
-
-  /**
-   * Remove the current registration cleanly (implementation defined cleanup)
-   * 
-   * @throws IOException
-   */
-  public void unregister() throws IOException;
-
-  /**
-   * Client API to get the list of instances registered via the current registry key.
-   * 
-   * @param component
-   * @return
-   * @throws IOException
-   */
-  public ServiceInstanceSet getInstances(String component) throws IOException;
-}

http://git-wip-us.apache.org/repos/asf/hive/blob/215c5770/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/registry/impl/LlapFixedRegistryImpl.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/registry/impl/LlapFixedRegistryImpl.java b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/registry/impl/LlapFixedRegistryImpl.java
deleted file mode 100644
index 621a6a6..0000000
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/registry/impl/LlapFixedRegistryImpl.java
+++ /dev/null
@@ -1,223 +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.daemon.registry.impl;
-
-import java.io.IOException;
-import java.net.InetAddress;
-import java.net.InetSocketAddress;
-import java.net.UnknownHostException;
-import java.util.HashMap;
-import java.util.HashSet;
-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.llap.configuration.LlapConfiguration;
-import org.apache.hadoop.hive.llap.daemon.registry.ServiceInstance;
-import org.apache.hadoop.hive.llap.daemon.registry.ServiceInstanceSet;
-import org.apache.hadoop.hive.llap.daemon.registry.ServiceRegistry;
-import org.apache.hadoop.net.NetUtils;
-import org.apache.hadoop.util.StringUtils;
-import org.apache.hadoop.yarn.api.records.Resource;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class LlapFixedRegistryImpl implements ServiceRegistry {
-
-  private static final Logger LOG = LoggerFactory.getLogger(LlapFixedRegistryImpl.class);
-
-  @InterfaceAudience.Private
-  // This is primarily for testing to avoid the host lookup
-  public static final String FIXED_REGISTRY_RESOLVE_HOST_NAMES = "fixed.registry.resolve.host.names";
-
-  private final int port;
-  private final int shuffle;
-  private final String[] hosts;
-  private final int memory;
-  private final int vcores;
-  private final boolean resolveHosts;
-
-  private final Map<String, String> srv = new HashMap<String, String>();
-
-  public LlapFixedRegistryImpl(String hosts, Configuration conf) {
-    this.hosts = hosts.split(",");
-    this.port =
-        conf.getInt(LlapConfiguration.LLAP_DAEMON_RPC_PORT,
-            LlapConfiguration.LLAP_DAEMON_RPC_PORT_DEFAULT);
-    this.shuffle =
-        conf.getInt(LlapConfiguration.LLAP_DAEMON_YARN_SHUFFLE_PORT,
-            LlapConfiguration.LLAP_DAEMON_YARN_SHUFFLE_PORT_DEFAULT);
-    this.resolveHosts = conf.getBoolean(FIXED_REGISTRY_RESOLVE_HOST_NAMES, true);
-
-    for (Map.Entry<String, String> kv : conf) {
-      if (kv.getKey().startsWith(LlapConfiguration.LLAP_DAEMON_PREFIX)
-          || kv.getKey().startsWith("hive.llap.")
-          || kv.getKey().startsWith(LlapConfiguration.LLAP_PREFIX)) {
-        // TODO: read this somewhere useful, like the task scheduler
-        srv.put(kv.getKey(), kv.getValue());
-      }
-    }
-
-    this.memory =
-        conf.getInt(LlapConfiguration.LLAP_DAEMON_MEMORY_PER_INSTANCE_MB,
-            LlapConfiguration.LLAP_DAEMON_MEMORY_PER_INSTANCE_MB_DEFAULT);
-    this.vcores =
-        conf.getInt(LlapConfiguration.LLAP_DAEMON_NUM_EXECUTORS,
-            LlapConfiguration.LLAP_DAEMON_NUM_EXECUTORS_DEFAULT);
-  }
-
-  @Override
-  public void start() throws InterruptedException {
-    // nothing to start
-  }
-
-  @Override
-  public void stop() throws InterruptedException {
-    // nothing to stop
-  }
-
-  @Override
-  public void register() throws IOException {
-    // nothing to register
-  }
-
-  @Override
-  public void unregister() throws IOException {
-    // nothing to unregister
-  }
-
-  public static String getWorkerIdentity(String host) {
-    // trigger clean errors for anyone who mixes up identity with hosts
-    return "host-" + host;
-  }
-
-  private final class FixedServiceInstance implements ServiceInstance {
-
-    private final String host;
-
-    public FixedServiceInstance(String host) {
-      if (resolveHosts) {
-        try {
-          InetAddress inetAddress = InetAddress.getByName(host);
-          if (NetUtils.isLocalAddress(inetAddress)) {
-            InetSocketAddress socketAddress = new InetSocketAddress(0);
-            socketAddress = NetUtils.getConnectAddress(socketAddress);
-            LOG.info("Adding host identified as local: " + host + " as "
-                + socketAddress.getHostName());
-            host = socketAddress.getHostName();
-          }
-        } catch (UnknownHostException e) {
-          LOG.warn("Ignoring resolution issues for host: " + host, e);
-        }
-      }
-      this.host = host;
-    }
-
-    @Override
-    public String getWorkerIdentity() {
-      return LlapFixedRegistryImpl.getWorkerIdentity(host);
-    }
-
-    @Override
-    public String getHost() {
-      return host;
-    }
-
-    @Override
-    public int getRpcPort() {
-      // TODO: allow >1 port per host?
-      return LlapFixedRegistryImpl.this.port;
-    }
-
-    @Override
-    public int getShufflePort() {
-      return LlapFixedRegistryImpl.this.shuffle;
-    }
-
-    @Override
-    public boolean isAlive() {
-      return true;
-    }
-
-    @Override
-    public Map<String, String> getProperties() {
-      Map<String, String> properties = new HashMap<>(srv);
-      // no worker identity
-      return properties;
-    }
-
-    @Override
-    public Resource getResource() {
-      return Resource.newInstance(memory, vcores);
-    }
-
-    @Override
-    public String toString() {
-      return "FixedServiceInstance{" +
-          "host=" + host +
-          ", memory=" + memory +
-          ", vcores=" + vcores +
-          '}';
-    }
-  }
-
-  private final class FixedServiceInstanceSet implements ServiceInstanceSet {
-
-    private final Map<String, ServiceInstance> instances = new HashMap<String, ServiceInstance>();
-
-    public FixedServiceInstanceSet() {
-      for (String host : hosts) {
-        // trigger bugs in anyone who uses this as a hostname
-        instances.put(getWorkerIdentity(host), new FixedServiceInstance(host));
-      }
-    }
-
-    @Override
-    public Map<String, ServiceInstance> getAll() {
-      return instances;
-    }
-
-    @Override
-    public ServiceInstance 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));
-      if (inst != null) {
-        byHost.add(inst);
-      }
-      return byHost;
-    }
-
-    @Override
-    public void refresh() throws IOException {
-      // I will do no such thing
-    }
-
-  }
-
-  @Override
-  public ServiceInstanceSet getInstances(String component) throws IOException {
-    return new FixedServiceInstanceSet();
-  }
-
-  @Override
-  public String toString() {
-    return String.format("FixedRegistry hosts=%s", StringUtils.join(",", this.hosts));
-  }
-}

http://git-wip-us.apache.org/repos/asf/hive/blob/215c5770/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/registry/impl/LlapRegistryService.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/registry/impl/LlapRegistryService.java b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/registry/impl/LlapRegistryService.java
deleted file mode 100644
index 6550940..0000000
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/registry/impl/LlapRegistryService.java
+++ /dev/null
@@ -1,87 +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.daemon.registry.impl;
-
-import java.io.IOException;
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hive.llap.configuration.LlapConfiguration;
-import org.apache.hadoop.hive.llap.daemon.registry.ServiceInstanceSet;
-import org.apache.hadoop.hive.llap.daemon.registry.ServiceRegistry;
-import org.apache.hadoop.service.AbstractService;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class LlapRegistryService extends AbstractService {
-
-  private static final Logger LOG = LoggerFactory.getLogger(LlapRegistryService.class);
-
-  private ServiceRegistry registry = null;
-  private final boolean isDaemon;
-
-  public LlapRegistryService(boolean isDaemon) {
-    super("LlapRegistryService");
-    this.isDaemon = isDaemon;
-  }
-
-  @Override
-  public void serviceInit(Configuration conf) {
-    String hosts = conf.getTrimmed(LlapConfiguration.LLAP_DAEMON_SERVICE_HOSTS);
-    if (hosts.startsWith("@")) {
-      registry = new LlapYarnRegistryImpl(hosts.substring(1), conf, isDaemon);
-    } else {
-      registry = new LlapFixedRegistryImpl(hosts, conf);
-    }
-    LOG.info("Using LLAP registry type " + registry);
-  }
-
-
-  @Override
-  public void serviceStart() throws Exception {
-    if (this.registry != null) {
-      this.registry.start();
-    }
-    if (isDaemon) {
-      registerWorker();
-    }
-  }
-
-  @Override
-  public void serviceStop() throws Exception {
-    if (isDaemon) {
-      unregisterWorker();
-    }
-    if (this.registry != null) {
-      this.registry.stop();
-    } else {
-      LOG.warn("Stopping non-existent registry service");
-    }
-  }
-
-  private void registerWorker() throws IOException {
-    if (this.registry != null) {
-      this.registry.register();
-    }
-  }
-
-  private void unregisterWorker() throws IOException {
-    if (this.registry != null) {
-      this.registry.unregister();
-    }
-  }
-
-  public ServiceInstanceSet getInstances() throws IOException {
-    return this.registry.getInstances("LLAP");
-  }
-}

http://git-wip-us.apache.org/repos/asf/hive/blob/215c5770/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/registry/impl/LlapYarnRegistryImpl.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/registry/impl/LlapYarnRegistryImpl.java b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/registry/impl/LlapYarnRegistryImpl.java
deleted file mode 100644
index 599da13..0000000
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/registry/impl/LlapYarnRegistryImpl.java
+++ /dev/null
@@ -1,384 +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.daemon.registry.impl;
-
-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.HashMap;
-import java.util.HashSet;
-import java.util.LinkedHashMap;
-import java.util.Map;
-import java.util.Set;
-import java.util.UUID;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
-
-import com.google.common.util.concurrent.ThreadFactoryBuilder;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hive.llap.configuration.LlapConfiguration;
-import org.apache.hadoop.hive.llap.daemon.registry.ServiceInstance;
-import org.apache.hadoop.hive.llap.daemon.registry.ServiceInstanceSet;
-import org.apache.hadoop.hive.llap.daemon.registry.ServiceRegistry;
-import org.apache.hadoop.registry.client.api.RegistryOperationsFactory;
-import org.apache.hadoop.registry.client.binding.RegistryPathUtils;
-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.impl.zk.RegistryOperationsService;
-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.yarn.api.records.Resource;
-import org.apache.hadoop.yarn.conf.YarnConfiguration;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.apache.tez.dag.api.TezUncheckedException;
-import org.apache.zookeeper.CreateMode;
-
-import com.google.common.base.Preconditions;
-
-public class LlapYarnRegistryImpl implements ServiceRegistry {
-
-  private static final Logger LOG = LoggerFactory.getLogger(LlapYarnRegistryImpl.class);
-
-  private final RegistryOperationsService client;
-  private final Configuration conf;
-  private final ServiceRecordMarshal encoder;
-  private final String path;
-
-  private final DynamicServiceInstanceSet instances = new DynamicServiceInstanceSet();
-
-  private static final UUID uniq = UUID.randomUUID();
-  private static final String hostname;
-
-  private static final String UNIQUE_IDENTIFIER = "llap.unique.id";
-
-  private final static String SERVICE_CLASS = "org-apache-hive";
-
-  final ScheduledExecutorService refresher = Executors.newScheduledThreadPool(1,
-      new ThreadFactoryBuilder().setDaemon(true).setNameFormat("LlapYarnRegistryRefresher").build());
-  final long refreshDelay;
-  private final boolean isDaemon;
-
-  static {
-    String localhost = "localhost";
-    try {
-      localhost = InetAddress.getLocalHost().getCanonicalHostName();
-    } catch (UnknownHostException uhe) {
-      // ignore
-    }
-    hostname = localhost;
-  }
-
-  public LlapYarnRegistryImpl(String instanceName, Configuration conf, boolean isDaemon) {
-
-    LOG.info("Llap Registry is enabled with registryid: " + instanceName);
-    this.conf = new Configuration(conf);
-    conf.addResource(YarnConfiguration.YARN_SITE_CONFIGURATION_FILE);
-    // registry reference
-    client = (RegistryOperationsService) RegistryOperationsFactory.createInstance(conf);
-    encoder = new RegistryUtils.ServiceRecordMarshal();
-    this.path = RegistryPathUtils.join(RegistryUtils.componentPath(RegistryUtils.currentUser(),
-        SERVICE_CLASS, instanceName, "workers"), "worker-");
-    refreshDelay =
-        conf.getInt(LlapConfiguration.LLAP_DAEMON_SERVICE_REFRESH_INTERVAL,
-            LlapConfiguration.LLAP_DAEMON_SERVICE_REFRESH_INTERVAL_DEFAULT);
-    this.isDaemon = isDaemon;
-    Preconditions.checkArgument(refreshDelay > 0,
-        "Refresh delay for registry has to be positive = %d", refreshDelay);
-  }
-
-  public Endpoint getRpcEndpoint() {
-    final int rpcPort =
-        conf.getInt(LlapConfiguration.LLAP_DAEMON_RPC_PORT,
-            LlapConfiguration.LLAP_DAEMON_RPC_PORT_DEFAULT);
-    return RegistryTypeUtils.ipcEndpoint("llap", new InetSocketAddress(hostname, rpcPort));
-  }
-
-  public Endpoint getShuffleEndpoint() {
-    final int shufflePort =
-        conf.getInt(LlapConfiguration.LLAP_DAEMON_YARN_SHUFFLE_PORT,
-            LlapConfiguration.LLAP_DAEMON_YARN_SHUFFLE_PORT_DEFAULT);
-    // HTTP today, but might not be
-    return RegistryTypeUtils.inetAddrEndpoint("shuffle", ProtocolTypes.PROTOCOL_TCP, hostname,
-        shufflePort);
-  }
-
-  public Endpoint getServicesEndpoint() {
-    final int servicePort =
-        conf.getInt(LlapConfiguration.LLAP_DAEMON_SERVICE_PORT,
-            LlapConfiguration.LLAP_DAEMON_SERVICE_PORT_DEFAULT);
-    final boolean isSSL =
-        conf.getBoolean(LlapConfiguration.LLAP_DAEMON_SERVICE_SSL,
-            LlapConfiguration.LLAP_DAEMON_SERVICE_SSL_DEFAULT);
-    final String scheme = isSSL ? "https" : "http";
-    final URL serviceURL;
-    try {
-      serviceURL = new URL(scheme, hostname, servicePort, "");
-      return RegistryTypeUtils.webEndpoint("services", serviceURL.toURI());
-    } catch (MalformedURLException e) {
-      throw new TezUncheckedException(e);
-    } catch (URISyntaxException e) {
-      throw new TezUncheckedException("llap service URI for " + hostname + " is invalid", e);
-    }
-  }
-
-  private final String getPath() {
-    return this.path;
-  }
-
-  @Override
-  public void register() throws IOException {
-    String path = getPath();
-    ServiceRecord srv = new ServiceRecord();
-    srv.addInternalEndpoint(getRpcEndpoint());
-    srv.addInternalEndpoint(getShuffleEndpoint());
-    srv.addExternalEndpoint(getServicesEndpoint());
-
-    for (Map.Entry<String, String> kv : this.conf) {
-      if (kv.getKey().startsWith(LlapConfiguration.LLAP_DAEMON_PREFIX)
-          || kv.getKey().startsWith("hive.llap.")) {
-        // TODO: read this somewhere useful, like the task scheduler
-        srv.set(kv.getKey(), kv.getValue());
-      }
-    }
-
-    // restart sensitive instance id
-    srv.set(UNIQUE_IDENTIFIER, uniq.toString());
-
-    client.mknode(RegistryPathUtils.parentOf(path), true);
-
-    // FIXME: YARN registry needs to expose Ephemeral_Seq nodes & return the paths
-    client.zkCreate(path, CreateMode.EPHEMERAL_SEQUENTIAL, encoder.toBytes(srv),
-        client.getClientAcls());
-  }
-
-  @Override
-  public void unregister() throws IOException {
-   // Nothing for the zkCreate models
-  }
-
-  private class DynamicServiceInstance implements ServiceInstance {
-
-    private final ServiceRecord srv;
-    private boolean alive = true;
-    private final String host;
-    private final int rpcPort;
-    private final int shufflePort;
-
-    public DynamicServiceInstance(ServiceRecord srv) throws IOException {
-      this.srv = srv;
-
-      final Endpoint shuffle = srv.getInternalEndpoint("shuffle");
-      final Endpoint rpc = srv.getInternalEndpoint("llap");
-
-      this.host =
-          RegistryTypeUtils.getAddressField(rpc.addresses.get(0),
-              AddressTypes.ADDRESS_HOSTNAME_FIELD);
-      this.rpcPort =
-          Integer.valueOf(RegistryTypeUtils.getAddressField(rpc.addresses.get(0),
-              AddressTypes.ADDRESS_PORT_FIELD));
-      this.shufflePort =
-          Integer.valueOf(RegistryTypeUtils.getAddressField(shuffle.addresses.get(0),
-              AddressTypes.ADDRESS_PORT_FIELD));
-    }
-
-    @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;
-    }
-
-    @Override
-    public boolean isAlive() {
-      return alive ;
-    }
-
-    public void kill() {
-      // May be possible to generate a notification back to the scheduler from here.
-      LOG.info("Killing service instance: " + this);
-      this.alive = false;
-    }
-
-    @Override
-    public Map<String, String> getProperties() {
-      return srv.attributes();
-    }
-
-    @Override
-    public Resource getResource() {
-      int memory = Integer.valueOf(srv.get(LlapConfiguration.LLAP_DAEMON_MEMORY_PER_INSTANCE_MB));
-      int vCores = Integer.valueOf(srv.get(LlapConfiguration.LLAP_DAEMON_NUM_EXECUTORS));
-      return Resource.newInstance(memory, vCores);
-    }
-
-    @Override
-    public String toString() {
-      return "DynamicServiceInstance [alive=" + alive + ", host=" + host + ":" + rpcPort + " with resources=" + getResource() +"]";
-    }
-
-    // Relying on the identity hashCode and equality, since refreshing instances retains the old copy
-    // of an already known instance.
-  }
-
-  private class DynamicServiceInstanceSet implements ServiceInstanceSet {
-
-    // LinkedHashMap to retain iteration order.
-    private final Map<String, ServiceInstance> instances = new LinkedHashMap<>();
-
-    @Override
-    public synchronized Map<String, ServiceInstance> getAll() {
-      // Return a copy. Instances may be modified during a refresh.
-      return new LinkedHashMap<>(instances);
-    }
-
-    @Override
-    public synchronized ServiceInstance getInstance(String name) {
-      return instances.get(name);
-    }
-
-    @Override
-    public  void refresh() throws IOException {
-      /* call this from wherever */
-      Map<String, ServiceInstance> freshInstances = new HashMap<String, ServiceInstance>();
-
-      String path = getPath();
-      Map<String, ServiceRecord> records =
-          RegistryUtils.listServiceRecords(client, RegistryPathUtils.parentOf(path));
-      // Synchronize after reading the service records from the external service (ZK)
-      synchronized (this) {
-        Set<String> latestKeys = new HashSet<String>();
-        LOG.info("Starting to refresh ServiceInstanceSet " + System.identityHashCode(this));
-        for (ServiceRecord rec : records.values()) {
-          ServiceInstance instance = new DynamicServiceInstance(rec);
-          if (instance != null) {
-            if (instances != null && instances.containsKey(instance.getWorkerIdentity()) == false) {
-              // add a new object
-              freshInstances.put(instance.getWorkerIdentity(), instance);
-              if (LOG.isInfoEnabled()) {
-                LOG.info("Adding new worker " + instance.getWorkerIdentity() + " which mapped to "
-                    + instance);
-              }
-            } else {
-              if (LOG.isDebugEnabled()) {
-                LOG.debug("Retaining running worker " + instance.getWorkerIdentity() +
-                    " which mapped to " + instance);
-              }
-            }
-          }
-          latestKeys.add(instance.getWorkerIdentity());
-        }
-
-        if (instances != null) {
-          // deep-copy before modifying
-          Set<String> oldKeys = new HashSet<>(instances.keySet());
-          if (oldKeys.removeAll(latestKeys)) {
-            // This is all the records which have not checked in, and are effectively dead.
-            for (String k : oldKeys) {
-              // this is so that people can hold onto ServiceInstance references as placeholders for tasks
-              final DynamicServiceInstance dead = (DynamicServiceInstance) instances.get(k);
-              dead.kill();
-              if (LOG.isInfoEnabled()) {
-                LOG.info("Deleting dead worker " + k + " which mapped to " + dead);
-              }
-            }
-          }
-          // oldKeys contains the set of dead instances at this point.
-          this.instances.keySet().removeAll(oldKeys);
-          this.instances.putAll(freshInstances);
-        } else {
-          this.instances.putAll(freshInstances);
-        }
-      }
-    }
-
-    @Override
-    public synchronized Set<ServiceInstance> getByHost(String host) {
-      // TODO Maybe store this as a map which is populated during construction, to avoid walking
-      // the map on each request.
-      Set<ServiceInstance> byHost = new HashSet<ServiceInstance>();
-
-      for (ServiceInstance i : instances.values()) {
-        if (host.equals(i.getHost())) {
-          // all hosts in instances should be alive in this impl
-          byHost.add(i);
-        }
-        if (LOG.isDebugEnabled()) {
-          LOG.debug("Locality comparing " + host + " to " + i.getHost());
-        }
-      }
-      if (LOG.isDebugEnabled()) {
-        LOG.debug("Returning " + byHost.size() + " hosts for locality allocation on " + host);
-      }
-      return byHost;
-    }
-  }
-
-  @Override
-  public ServiceInstanceSet getInstances(String component) throws IOException {
-    Preconditions.checkArgument("LLAP".equals(component)); // right now there is only 1 component 
-    if (this.client != null) {
-      instances.refresh();
-      return instances;
-    } else {
-      Preconditions.checkNotNull(this.client, "Yarn registry client is not intialized");
-      return null;
-    }
-  }
-
-  @Override
-  public void start() {
-    if (client == null) return;
-    client.start();
-    if (isDaemon) return;
-    refresher.scheduleWithFixedDelay(new Runnable() {
-      @Override
-      public void run() {
-        try {
-          instances.refresh();
-        } catch (IOException ioe) {
-          LOG.warn("Could not refresh hosts during scheduled refresh", ioe);
-        }
-      }
-    }, 0, refreshDelay, TimeUnit.SECONDS);
-  }
-
-  @Override
-  public void stop() {
-    if (client != null) {
-      client.stop();
-    }
-  }
-}

http://git-wip-us.apache.org/repos/asf/hive/blob/215c5770/llap-server/src/java/org/apache/tez/dag/app/rm/LlapTaskSchedulerService.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/tez/dag/app/rm/LlapTaskSchedulerService.java b/llap-server/src/java/org/apache/tez/dag/app/rm/LlapTaskSchedulerService.java
index 7fb9a99..6fd01f9 100644
--- a/llap-server/src/java/org/apache/tez/dag/app/rm/LlapTaskSchedulerService.java
+++ b/llap-server/src/java/org/apache/tez/dag/app/rm/LlapTaskSchedulerService.java
@@ -51,9 +51,9 @@ import com.google.common.util.concurrent.FutureCallback;
 import com.google.common.util.concurrent.Futures;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hive.llap.configuration.LlapConfiguration;
-import org.apache.hadoop.hive.llap.daemon.registry.ServiceInstance;
-import org.apache.hadoop.hive.llap.daemon.registry.ServiceInstanceSet;
-import org.apache.hadoop.hive.llap.daemon.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.impl.LlapRegistryService;
 import org.apache.hadoop.yarn.api.records.Container;
 import org.apache.hadoop.yarn.api.records.ContainerId;
 import org.apache.hadoop.yarn.api.records.NodeId;

http://git-wip-us.apache.org/repos/asf/hive/blob/215c5770/llap-server/src/test/org/apache/tez/dag/app/rm/TestLlapTaskSchedulerService.java
----------------------------------------------------------------------
diff --git a/llap-server/src/test/org/apache/tez/dag/app/rm/TestLlapTaskSchedulerService.java b/llap-server/src/test/org/apache/tez/dag/app/rm/TestLlapTaskSchedulerService.java
index 2f93266..23724a4 100644
--- a/llap-server/src/test/org/apache/tez/dag/app/rm/TestLlapTaskSchedulerService.java
+++ b/llap-server/src/test/org/apache/tez/dag/app/rm/TestLlapTaskSchedulerService.java
@@ -34,7 +34,7 @@ import java.util.concurrent.locks.ReentrantLock;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hive.llap.configuration.LlapConfiguration;
-import org.apache.hadoop.hive.llap.daemon.registry.impl.LlapFixedRegistryImpl;
+import org.apache.hadoop.hive.llap.registry.impl.LlapFixedRegistryImpl;
 import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
 import org.apache.hadoop.yarn.api.records.ApplicationId;
 import org.apache.hadoop.yarn.api.records.Container;


Mime
View raw message