You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by ss...@apache.org on 2016/02/01 18:58:05 UTC
[06/11] hive git commit: HIVE-12934. Refactor llap module structure
to allow for a usable client. (Siddharth Seth,
reviewed by Sergey Shelukhin) (cherry picked from commit
4185d9b8e2eecfef3b5a38899f6928fa82c01e99)
http://git-wip-us.apache.org/repos/asf/hive/blob/9886414b/llap-server/src/java/org/apache/hadoop/hive/llap/LlapNodeId.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/LlapNodeId.java b/llap-server/src/java/org/apache/hadoop/hive/llap/LlapNodeId.java
deleted file mode 100644
index 515bf3c..0000000
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/LlapNodeId.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;
-
-import com.google.common.cache.CacheBuilder;
-import com.google.common.cache.CacheLoader;
-import com.google.common.cache.LoadingCache;
-
-public class LlapNodeId {
-
- private static final LoadingCache<LlapNodeId, LlapNodeId> CACHE =
- CacheBuilder.newBuilder().softValues().build(
- new CacheLoader<LlapNodeId, LlapNodeId>() {
- @Override
- public LlapNodeId load(LlapNodeId key) throws Exception {
- return key;
- }
- });
-
- public static LlapNodeId getInstance(String hostname, int port) {
- return CACHE.getUnchecked(new LlapNodeId(hostname, port));
- }
-
-
- private final String hostname;
- private final int port;
-
-
- private LlapNodeId(String hostname, int port) {
- this.hostname = hostname;
- this.port = port;
- }
-
- public String getHostname() {
- return hostname;
- }
-
- public int getPort() {
- return port;
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
-
- LlapNodeId that = (LlapNodeId) o;
-
- if (port != that.port) {
- return false;
- }
- if (!hostname.equals(that.hostname)) {
- return false;
- }
-
- return true;
- }
-
- @Override
- public int hashCode() {
- int result = hostname.hashCode();
- result = 1009 * result + port;
- return result;
- }
-
- @Override
- public String toString() {
- return hostname + ":" + port;
- }
-}
http://git-wip-us.apache.org/repos/asf/hive/blob/9886414b/llap-server/src/java/org/apache/hadoop/hive/llap/cli/LlapServiceDriver.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/cli/LlapServiceDriver.java b/llap-server/src/java/org/apache/hadoop/hive/llap/cli/LlapServiceDriver.java
index 544af09..0399798 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/cli/LlapServiceDriver.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/cli/LlapServiceDriver.java
@@ -25,6 +25,8 @@ import java.util.Map;
import java.util.Map.Entry;
import java.util.Collection;
+import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos;
+import org.apache.hadoop.hive.llap.tezplugins.LlapTezUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.hadoop.conf.Configuration;
@@ -217,7 +219,13 @@ public class LlapServiceDriver {
CompressionUtils.unTar(new Path(libDir, "tez.tar.gz").toString(), libDir.toString(), true);
lfs.delete(new Path(libDir, "tez.tar.gz"), false);
+ // llap-common
+ lfs.copyFromLocalFile(new Path(Utilities.jarFinderGetJar(LlapDaemonProtocolProtos.class)), libDir);
+ // llap-tez
+ lfs.copyFromLocalFile(new Path(Utilities.jarFinderGetJar(LlapTezUtils.class)), libDir);
+ // llap-server
lfs.copyFromLocalFile(new Path(Utilities.jarFinderGetJar(LlapInputFormat.class)), libDir);
+ // hive-exec
lfs.copyFromLocalFile(new Path(Utilities.jarFinderGetJar(HiveInputFormat.class)), libDir);
// copy default aux classes (json/hbase)
http://git-wip-us.apache.org/repos/asf/hive/blob/9886414b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/LlapDaemonProtocolBlockingPB.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/LlapDaemonProtocolBlockingPB.java b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/LlapDaemonProtocolBlockingPB.java
deleted file mode 100644
index 4c09941..0000000
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/LlapDaemonProtocolBlockingPB.java
+++ /dev/null
@@ -1,28 +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;
-
-import org.apache.hadoop.ipc.ProtocolInfo;
-import org.apache.hadoop.security.KerberosInfo;
-import org.apache.hadoop.security.token.TokenInfo;
-import org.apache.hadoop.hive.conf.HiveConf;
-import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos;
-import org.apache.hadoop.hive.llap.security.LlapTokenSelector;
-
-@ProtocolInfo(protocolName = "org.apache.hadoop.hive.llap.daemon.LlapDaemonProtocolBlockingPB", protocolVersion = 1)
-@KerberosInfo(serverPrincipal = HiveConf.HIVE_LLAP_DAEMON_SERVICE_PRINCIPAL_NAME)
-@TokenInfo(LlapTokenSelector.class)
-public interface LlapDaemonProtocolBlockingPB extends LlapDaemonProtocolProtos.LlapDaemonProtocol.BlockingInterface {
-}
http://git-wip-us.apache.org/repos/asf/hive/blob/9886414b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/LlapManagementProtocolBlockingPB.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/LlapManagementProtocolBlockingPB.java b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/LlapManagementProtocolBlockingPB.java
deleted file mode 100644
index 4efadac..0000000
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/LlapManagementProtocolBlockingPB.java
+++ /dev/null
@@ -1,24 +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;
-
-import org.apache.hadoop.hive.conf.HiveConf;
-import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos;
-import org.apache.hadoop.ipc.ProtocolInfo;
-import org.apache.hadoop.security.KerberosInfo;
-
-@ProtocolInfo(protocolName = "org.apache.hadoop.hive.llap.daemon.LlapManagementProtocolBlockingPB", protocolVersion = 1)
-@KerberosInfo(serverPrincipal = HiveConf.HIVE_LLAP_DAEMON_SERVICE_PRINCIPAL_NAME)
-public interface LlapManagementProtocolBlockingPB extends LlapDaemonProtocolProtos.LlapManagementProtocol.BlockingInterface {
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hive/blob/9886414b/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 98951e1..37120d2 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
@@ -65,7 +65,7 @@ public class LlapDaemon extends CompositeService implements ContainerRunner, Lla
private static final Logger LOG = LoggerFactory.getLogger(LlapDaemon.class);
private final Configuration shuffleHandlerConf;
- private final LlapDaemonProtocolServerImpl server;
+ private final LlapProtocolServerImpl server;
private final ContainerRunnerImpl containerRunner;
private final AMReporter amReporter;
private final LlapRegistryService registry;
@@ -164,7 +164,7 @@ public class LlapDaemon extends CompositeService implements ContainerRunner, Lla
this.amReporter = new AMReporter(srvAddress, new QueryFailedHandlerProxy(), daemonConf);
- this.server = new LlapDaemonProtocolServerImpl(
+ this.server = new LlapProtocolServerImpl(
numHandlers, this, srvAddress, mngAddress, srvPort, mngPort);
this.containerRunner = new ContainerRunnerImpl(daemonConf,
http://git-wip-us.apache.org/repos/asf/hive/blob/9886414b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/LlapDaemonProtocolClientImpl.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/LlapDaemonProtocolClientImpl.java b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/LlapDaemonProtocolClientImpl.java
deleted file mode 100644
index 9c7d2e2..0000000
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/LlapDaemonProtocolClientImpl.java
+++ /dev/null
@@ -1,125 +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.impl;
-
-import javax.annotation.Nullable;
-import javax.net.SocketFactory;
-import java.io.IOException;
-import java.net.InetSocketAddress;
-
-import com.google.protobuf.RpcController;
-import com.google.protobuf.ServiceException;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.QueryCompleteRequestProto;
-import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.QueryCompleteResponseProto;
-import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.SourceStateUpdatedRequestProto;
-import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.SourceStateUpdatedResponseProto;
-import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.SubmitWorkRequestProto;
-import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.SubmitWorkResponseProto;
-import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.TerminateFragmentRequestProto;
-import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.TerminateFragmentResponseProto;
-import org.apache.hadoop.io.retry.RetryPolicy;
-import org.apache.hadoop.ipc.ProtobufRpcEngine;
-import org.apache.hadoop.ipc.ProtocolProxy;
-import org.apache.hadoop.ipc.RPC;
-import org.apache.hadoop.net.NetUtils;
-import org.apache.hadoop.hive.llap.daemon.LlapDaemonProtocolBlockingPB;
-import org.apache.hadoop.security.UserGroupInformation;
-
-// TODO Change all this to be based on a regular interface instead of relying on the Proto service - Exception signatures cannot be controlled without this for the moment.
-
-
-public class LlapDaemonProtocolClientImpl implements LlapDaemonProtocolBlockingPB {
-
- private final Configuration conf;
- private final InetSocketAddress serverAddr;
- private final RetryPolicy retryPolicy;
- private final SocketFactory socketFactory;
- LlapDaemonProtocolBlockingPB proxy;
-
-
- public LlapDaemonProtocolClientImpl(Configuration conf, String hostname, int port,
- @Nullable RetryPolicy retryPolicy,
- @Nullable SocketFactory socketFactory) {
- this.conf = conf;
- this.serverAddr = NetUtils.createSocketAddr(hostname, port);
- this.retryPolicy = retryPolicy;
- if (socketFactory == null) {
- this.socketFactory = NetUtils.getDefaultSocketFactory(conf);
- } else {
- this.socketFactory = socketFactory;
- }
- }
-
- @Override
- public SubmitWorkResponseProto submitWork(RpcController controller,
- SubmitWorkRequestProto request) throws
- ServiceException {
- try {
- return getProxy().submitWork(null, request);
- } catch (IOException e) {
- throw new ServiceException(e);
- }
- }
-
- @Override
- public SourceStateUpdatedResponseProto sourceStateUpdated(RpcController controller,
- SourceStateUpdatedRequestProto request) throws
- ServiceException {
- try {
- return getProxy().sourceStateUpdated(null, request);
- } catch (IOException e) {
- throw new ServiceException(e);
- }
- }
-
- @Override
- public QueryCompleteResponseProto queryComplete(RpcController controller,
- QueryCompleteRequestProto request) throws
- ServiceException {
- try {
- return getProxy().queryComplete(null, request);
- } catch (IOException e) {
- throw new ServiceException(e);
- }
- }
-
- @Override
- public TerminateFragmentResponseProto terminateFragment(
- RpcController controller,
- TerminateFragmentRequestProto request) throws ServiceException {
- try {
- return getProxy().terminateFragment(null, request);
- } catch (IOException e) {
- throw new ServiceException(e);
- }
- }
-
- public LlapDaemonProtocolBlockingPB getProxy() throws IOException {
- if (proxy == null) {
- proxy = createProxy();
- }
- return proxy;
- }
-
- public LlapDaemonProtocolBlockingPB createProxy() throws IOException {
- RPC.setProtocolEngine(conf, LlapDaemonProtocolBlockingPB.class, ProtobufRpcEngine.class);
- ProtocolProxy<LlapDaemonProtocolBlockingPB> proxy =
- RPC.getProtocolProxy(LlapDaemonProtocolBlockingPB.class, 0, serverAddr,
- UserGroupInformation.getCurrentUser(), conf, NetUtils.getDefaultSocketFactory(conf), 0,
- retryPolicy);
- return proxy.getProxy();
- }
-}
http://git-wip-us.apache.org/repos/asf/hive/blob/9886414b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/LlapDaemonProtocolServerImpl.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/LlapDaemonProtocolServerImpl.java b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/LlapDaemonProtocolServerImpl.java
deleted file mode 100644
index 45ca906..0000000
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/LlapDaemonProtocolServerImpl.java
+++ /dev/null
@@ -1,251 +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.impl;
-
-import java.io.IOException;
-import java.net.InetSocketAddress;
-import java.security.PrivilegedAction;
-import java.util.concurrent.atomic.AtomicReference;
-
-import com.google.common.annotations.VisibleForTesting;
-import com.google.common.io.ByteArrayDataOutput;
-import com.google.common.io.ByteStreams;
-import com.google.protobuf.BlockingService;
-import com.google.protobuf.ByteString;
-import com.google.protobuf.RpcController;
-import com.google.protobuf.ServiceException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.apache.hadoop.classification.InterfaceAudience;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
-import org.apache.hadoop.hive.conf.HiveConf;
-import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
-import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos;
-import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.GetTokenRequestProto;
-import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.GetTokenResponseProto;
-import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.QueryCompleteRequestProto;
-import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.QueryCompleteResponseProto;
-import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.SourceStateUpdatedRequestProto;
-import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.SourceStateUpdatedResponseProto;
-import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.SubmitWorkRequestProto;
-import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.SubmitWorkResponseProto;
-import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.TerminateFragmentRequestProto;
-import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.TerminateFragmentResponseProto;
-import org.apache.hadoop.io.Text;
-import org.apache.hadoop.ipc.ProtobufRpcEngine;
-import org.apache.hadoop.ipc.RPC;
-import org.apache.hadoop.net.NetUtils;
-import org.apache.hadoop.security.UserGroupInformation;
-import org.apache.hadoop.security.token.Token;
-import org.apache.hadoop.hive.llap.security.LlapSecurityHelper;
-import org.apache.hadoop.hive.llap.security.LlapTokenIdentifier;
-import org.apache.hadoop.hive.llap.security.SecretManager;
-import org.apache.hadoop.service.AbstractService;
-import org.apache.hadoop.hive.llap.daemon.ContainerRunner;
-import org.apache.hadoop.hive.llap.daemon.LlapDaemonProtocolBlockingPB;
-import org.apache.hadoop.hive.llap.daemon.LlapManagementProtocolBlockingPB;
-import org.apache.hadoop.hive.llap.security.LlapDaemonPolicyProvider;
-
-public class LlapDaemonProtocolServerImpl extends AbstractService
- implements LlapDaemonProtocolBlockingPB, LlapManagementProtocolBlockingPB {
-
- private static final Logger LOG = LoggerFactory.getLogger(LlapDaemonProtocolServerImpl.class);
-
- private final int numHandlers;
- private final ContainerRunner containerRunner;
- private final int srvPort, mngPort;
- private RPC.Server server, mngServer;
- private final AtomicReference<InetSocketAddress> srvAddress, mngAddress;
- private SecretManager zkSecretManager;
-
- public LlapDaemonProtocolServerImpl(int numHandlers,
- ContainerRunner containerRunner,
- AtomicReference<InetSocketAddress> srvAddress,
- AtomicReference<InetSocketAddress> mngAddress,
- int srvPort,
- int mngPort) {
- super("LlapDaemonProtocolServerImpl");
- this.numHandlers = numHandlers;
- this.containerRunner = containerRunner;
- this.srvAddress = srvAddress;
- this.srvPort = srvPort;
- this.mngAddress = mngAddress;
- this.mngPort = mngPort;
- LOG.info("Creating: " + LlapDaemonProtocolServerImpl.class.getSimpleName() +
- " with port configured to: " + srvPort);
- }
-
- @Override
- public SubmitWorkResponseProto submitWork(RpcController controller,
- SubmitWorkRequestProto request) throws
- ServiceException {
- try {
- return containerRunner.submitWork(request);
- } catch (IOException e) {
- throw new ServiceException(e);
- }
- }
-
- @Override
- public SourceStateUpdatedResponseProto sourceStateUpdated(RpcController controller,
- SourceStateUpdatedRequestProto request) throws ServiceException {
- return containerRunner.sourceStateUpdated(request);
- }
-
- @Override
- public QueryCompleteResponseProto queryComplete(RpcController controller,
- QueryCompleteRequestProto request) throws ServiceException {
- return containerRunner.queryComplete(request);
- }
-
- @Override
- public TerminateFragmentResponseProto terminateFragment(
- RpcController controller,
- TerminateFragmentRequestProto request) throws ServiceException {
- return containerRunner.terminateFragment(request);
- }
-
- @Override
- public void serviceStart() {
- final Configuration conf = getConfig();
- final BlockingService daemonImpl =
- LlapDaemonProtocolProtos.LlapDaemonProtocol.newReflectiveBlockingService(this);
- final BlockingService managementImpl =
- LlapDaemonProtocolProtos.LlapManagementProtocol.newReflectiveBlockingService(this);
- if (!UserGroupInformation.isSecurityEnabled()) {
- startProtocolServers(conf, daemonImpl, managementImpl);
- return;
- }
- String llapPrincipal = HiveConf.getVar(conf, ConfVars.LLAP_KERBEROS_PRINCIPAL),
- llapKeytab = HiveConf.getVar(conf, ConfVars.LLAP_KERBEROS_KEYTAB_FILE);
- zkSecretManager = SecretManager.createSecretManager(conf, llapPrincipal, llapKeytab);
-
- // Start the protocol server after properly authenticating with daemon keytab.
- UserGroupInformation daemonUgi = null;
- try {
- daemonUgi = LlapSecurityHelper.loginWithKerberos(llapPrincipal, llapKeytab);
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- daemonUgi.doAs(new PrivilegedAction<Void>() {
- @Override
- public Void run() {
- startProtocolServers(conf, daemonImpl, managementImpl);
- return null;
- }
- });
- }
-
- private void startProtocolServers(
- Configuration conf, BlockingService daemonImpl, BlockingService managementImpl) {
- server = startProtocolServer(srvPort, numHandlers, srvAddress, conf, daemonImpl,
- LlapDaemonProtocolBlockingPB.class);
- mngServer = startProtocolServer(mngPort, 2, mngAddress, conf, managementImpl,
- LlapManagementProtocolBlockingPB.class);
- }
-
- private RPC.Server startProtocolServer(int srvPort, int numHandlers,
- AtomicReference<InetSocketAddress> bindAddress, Configuration conf,
- BlockingService impl, Class<?> protocolClass) {
- InetSocketAddress addr = new InetSocketAddress(srvPort);
- RPC.Server server;
- try {
- server = createServer(protocolClass, addr, conf, numHandlers, impl);
- server.start();
- } catch (IOException e) {
- LOG.error("Failed to run RPC Server on port: " + srvPort, e);
- throw new RuntimeException(e);
- }
-
- InetSocketAddress serverBindAddress = NetUtils.getConnectAddress(server);
- bindAddress.set(NetUtils.createSocketAddrForHost(
- serverBindAddress.getAddress().getCanonicalHostName(),
- serverBindAddress.getPort()));
- LOG.info("Instantiated " + protocolClass.getSimpleName() + " at " + bindAddress);
- return server;
- }
-
- @Override
- public void serviceStop() {
- if (server != null) {
- server.stop();
- }
- if (mngServer != null) {
- mngServer.stop();
- }
- }
-
- @InterfaceAudience.Private
- @VisibleForTesting
- InetSocketAddress getBindAddress() {
- return srvAddress.get();
- }
-
- private RPC.Server createServer(Class<?> pbProtocol, InetSocketAddress addr, Configuration conf,
- int numHandlers, BlockingService blockingService) throws
- IOException {
- RPC.setProtocolEngine(conf, pbProtocol, ProtobufRpcEngine.class);
- RPC.Builder builder = new RPC.Builder(conf)
- .setProtocol(pbProtocol)
- .setInstance(blockingService)
- .setBindAddress(addr.getHostName())
- .setPort(addr.getPort())
- .setNumHandlers(numHandlers);
- if (zkSecretManager != null) {
- builder = builder.setSecretManager(zkSecretManager);
- }
- RPC.Server server = builder.build();
-
- if (conf.getBoolean(CommonConfigurationKeysPublic.HADOOP_SECURITY_AUTHORIZATION, false)) {
- server.refreshServiceAcl(conf, new LlapDaemonPolicyProvider());
- }
- return server;
- }
-
-
- @Override
- public GetTokenResponseProto getDelegationToken(RpcController controller,
- GetTokenRequestProto request) throws ServiceException {
- if (zkSecretManager == null) {
- throw new ServiceException("Operation not supported on unsecure cluster");
- }
- UserGroupInformation ugi;
- try {
- ugi = UserGroupInformation.getCurrentUser();
- } catch (IOException e) {
- throw new ServiceException(e);
- }
- String user = ugi.getUserName();
- Text owner = new Text(user);
- Text realUser = null;
- if (ugi.getRealUser() != null) {
- realUser = new Text(ugi.getRealUser().getUserName());
- }
- Text renewer = new Text(ugi.getShortUserName());
- LlapTokenIdentifier llapId = new LlapTokenIdentifier(owner, renewer, realUser);
- // TODO: note that the token is not renewable right now and will last for 2 weeks by default.
- Token<LlapTokenIdentifier> token = new Token<LlapTokenIdentifier>(llapId, zkSecretManager);
- ByteArrayDataOutput out = ByteStreams.newDataOutput();
- try {
- token.write(out);
- } catch (IOException e) {
- throw new ServiceException(e);
- }
- ByteString bs = ByteString.copyFrom(out.toByteArray());
- GetTokenResponseProto response = GetTokenResponseProto.newBuilder().setToken(bs).build();
- return response;
- }
-}
http://git-wip-us.apache.org/repos/asf/hive/blob/9886414b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/LlapManagementProtocolClientImpl.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/LlapManagementProtocolClientImpl.java b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/LlapManagementProtocolClientImpl.java
deleted file mode 100644
index e293a95..0000000
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/LlapManagementProtocolClientImpl.java
+++ /dev/null
@@ -1,82 +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.impl;
-
-import javax.annotation.Nullable;
-import javax.net.SocketFactory;
-import java.io.IOException;
-import java.net.InetSocketAddress;
-
-import com.google.protobuf.RpcController;
-import com.google.protobuf.ServiceException;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.io.retry.RetryPolicy;
-import org.apache.hadoop.ipc.ProtobufRpcEngine;
-import org.apache.hadoop.ipc.ProtocolProxy;
-import org.apache.hadoop.ipc.RPC;
-import org.apache.hadoop.net.NetUtils;
-import org.apache.hadoop.hive.llap.daemon.LlapManagementProtocolBlockingPB;
-import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.GetTokenRequestProto;
-import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.GetTokenResponseProto;
-import org.apache.hadoop.security.UserGroupInformation;
-
-public class LlapManagementProtocolClientImpl implements LlapManagementProtocolBlockingPB {
-
- private final Configuration conf;
- private final InetSocketAddress serverAddr;
- private final RetryPolicy retryPolicy;
- private final SocketFactory socketFactory;
- LlapManagementProtocolBlockingPB proxy;
-
-
- public LlapManagementProtocolClientImpl(Configuration conf, String hostname, int port,
- @Nullable RetryPolicy retryPolicy,
- @Nullable SocketFactory socketFactory) {
- this.conf = conf;
- this.serverAddr = NetUtils.createSocketAddr(hostname, port);
- this.retryPolicy = retryPolicy;
- if (socketFactory == null) {
- this.socketFactory = NetUtils.getDefaultSocketFactory(conf);
- } else {
- this.socketFactory = socketFactory;
- }
- }
-
- public LlapManagementProtocolBlockingPB getProxy() throws IOException {
- if (proxy == null) {
- proxy = createProxy();
- }
- return proxy;
- }
-
- public LlapManagementProtocolBlockingPB createProxy() throws IOException {
- RPC.setProtocolEngine(conf, LlapManagementProtocolBlockingPB.class, ProtobufRpcEngine.class);
- ProtocolProxy<LlapManagementProtocolBlockingPB> proxy =
- RPC.getProtocolProxy(LlapManagementProtocolBlockingPB.class, 0, serverAddr,
- UserGroupInformation.getCurrentUser(), conf, NetUtils.getDefaultSocketFactory(conf), 0,
- retryPolicy);
- return proxy.getProxy();
- }
-
- @Override
- public GetTokenResponseProto getDelegationToken(RpcController controller,
- GetTokenRequestProto request) throws ServiceException {
- try {
- return getProxy().getDelegationToken(null, request);
- } catch (IOException e) {
- throw new ServiceException(e);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/hive/blob/9886414b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/LlapProtocolServerImpl.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/LlapProtocolServerImpl.java b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/LlapProtocolServerImpl.java
new file mode 100644
index 0000000..c386d77
--- /dev/null
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/LlapProtocolServerImpl.java
@@ -0,0 +1,251 @@
+/*
+ * 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.impl;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.security.PrivilegedAction;
+import java.util.concurrent.atomic.AtomicReference;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.io.ByteArrayDataOutput;
+import com.google.common.io.ByteStreams;
+import com.google.protobuf.BlockingService;
+import com.google.protobuf.ByteString;
+import com.google.protobuf.RpcController;
+import com.google.protobuf.ServiceException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
+import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos;
+import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.GetTokenRequestProto;
+import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.GetTokenResponseProto;
+import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.QueryCompleteRequestProto;
+import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.QueryCompleteResponseProto;
+import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.SourceStateUpdatedRequestProto;
+import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.SourceStateUpdatedResponseProto;
+import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.SubmitWorkRequestProto;
+import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.SubmitWorkResponseProto;
+import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.TerminateFragmentRequestProto;
+import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.TerminateFragmentResponseProto;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.ipc.ProtobufRpcEngine;
+import org.apache.hadoop.ipc.RPC;
+import org.apache.hadoop.net.NetUtils;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.security.token.Token;
+import org.apache.hadoop.hive.llap.security.LlapSecurityHelper;
+import org.apache.hadoop.hive.llap.security.LlapTokenIdentifier;
+import org.apache.hadoop.hive.llap.security.SecretManager;
+import org.apache.hadoop.service.AbstractService;
+import org.apache.hadoop.hive.llap.daemon.ContainerRunner;
+import org.apache.hadoop.hive.llap.protocol.LlapProtocolBlockingPB;
+import org.apache.hadoop.hive.llap.protocol.LlapManagementProtocolPB;
+import org.apache.hadoop.hive.llap.security.LlapDaemonPolicyProvider;
+
+public class LlapProtocolServerImpl extends AbstractService
+ implements LlapProtocolBlockingPB, LlapManagementProtocolPB {
+
+ private static final Logger LOG = LoggerFactory.getLogger(LlapProtocolServerImpl.class);
+
+ private final int numHandlers;
+ private final ContainerRunner containerRunner;
+ private final int srvPort, mngPort;
+ private RPC.Server server, mngServer;
+ private final AtomicReference<InetSocketAddress> srvAddress, mngAddress;
+ private SecretManager zkSecretManager;
+
+ public LlapProtocolServerImpl(int numHandlers,
+ ContainerRunner containerRunner,
+ AtomicReference<InetSocketAddress> srvAddress,
+ AtomicReference<InetSocketAddress> mngAddress,
+ int srvPort,
+ int mngPort) {
+ super("LlapDaemonProtocolServerImpl");
+ this.numHandlers = numHandlers;
+ this.containerRunner = containerRunner;
+ this.srvAddress = srvAddress;
+ this.srvPort = srvPort;
+ this.mngAddress = mngAddress;
+ this.mngPort = mngPort;
+ LOG.info("Creating: " + LlapProtocolServerImpl.class.getSimpleName() +
+ " with port configured to: " + srvPort);
+ }
+
+ @Override
+ public SubmitWorkResponseProto submitWork(RpcController controller,
+ SubmitWorkRequestProto request) throws
+ ServiceException {
+ try {
+ return containerRunner.submitWork(request);
+ } catch (IOException e) {
+ throw new ServiceException(e);
+ }
+ }
+
+ @Override
+ public SourceStateUpdatedResponseProto sourceStateUpdated(RpcController controller,
+ SourceStateUpdatedRequestProto request) throws ServiceException {
+ return containerRunner.sourceStateUpdated(request);
+ }
+
+ @Override
+ public QueryCompleteResponseProto queryComplete(RpcController controller,
+ QueryCompleteRequestProto request) throws ServiceException {
+ return containerRunner.queryComplete(request);
+ }
+
+ @Override
+ public TerminateFragmentResponseProto terminateFragment(
+ RpcController controller,
+ TerminateFragmentRequestProto request) throws ServiceException {
+ return containerRunner.terminateFragment(request);
+ }
+
+ @Override
+ public void serviceStart() {
+ final Configuration conf = getConfig();
+ final BlockingService daemonImpl =
+ LlapDaemonProtocolProtos.LlapDaemonProtocol.newReflectiveBlockingService(this);
+ final BlockingService managementImpl =
+ LlapDaemonProtocolProtos.LlapManagementProtocol.newReflectiveBlockingService(this);
+ if (!UserGroupInformation.isSecurityEnabled()) {
+ startProtocolServers(conf, daemonImpl, managementImpl);
+ return;
+ }
+ String llapPrincipal = HiveConf.getVar(conf, ConfVars.LLAP_KERBEROS_PRINCIPAL),
+ llapKeytab = HiveConf.getVar(conf, ConfVars.LLAP_KERBEROS_KEYTAB_FILE);
+ zkSecretManager = SecretManager.createSecretManager(conf, llapPrincipal, llapKeytab);
+
+ // Start the protocol server after properly authenticating with daemon keytab.
+ UserGroupInformation daemonUgi = null;
+ try {
+ daemonUgi = LlapSecurityHelper.loginWithKerberos(llapPrincipal, llapKeytab);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ daemonUgi.doAs(new PrivilegedAction<Void>() {
+ @Override
+ public Void run() {
+ startProtocolServers(conf, daemonImpl, managementImpl);
+ return null;
+ }
+ });
+ }
+
+ private void startProtocolServers(
+ Configuration conf, BlockingService daemonImpl, BlockingService managementImpl) {
+ server = startProtocolServer(srvPort, numHandlers, srvAddress, conf, daemonImpl,
+ LlapProtocolBlockingPB.class);
+ mngServer = startProtocolServer(mngPort, 2, mngAddress, conf, managementImpl,
+ LlapManagementProtocolPB.class);
+ }
+
+ private RPC.Server startProtocolServer(int srvPort, int numHandlers,
+ AtomicReference<InetSocketAddress> bindAddress, Configuration conf,
+ BlockingService impl, Class<?> protocolClass) {
+ InetSocketAddress addr = new InetSocketAddress(srvPort);
+ RPC.Server server;
+ try {
+ server = createServer(protocolClass, addr, conf, numHandlers, impl);
+ server.start();
+ } catch (IOException e) {
+ LOG.error("Failed to run RPC Server on port: " + srvPort, e);
+ throw new RuntimeException(e);
+ }
+
+ InetSocketAddress serverBindAddress = NetUtils.getConnectAddress(server);
+ bindAddress.set(NetUtils.createSocketAddrForHost(
+ serverBindAddress.getAddress().getCanonicalHostName(),
+ serverBindAddress.getPort()));
+ LOG.info("Instantiated " + protocolClass.getSimpleName() + " at " + bindAddress);
+ return server;
+ }
+
+ @Override
+ public void serviceStop() {
+ if (server != null) {
+ server.stop();
+ }
+ if (mngServer != null) {
+ mngServer.stop();
+ }
+ }
+
+ @InterfaceAudience.Private
+ @VisibleForTesting
+ InetSocketAddress getBindAddress() {
+ return srvAddress.get();
+ }
+
+ private RPC.Server createServer(Class<?> pbProtocol, InetSocketAddress addr, Configuration conf,
+ int numHandlers, BlockingService blockingService) throws
+ IOException {
+ RPC.setProtocolEngine(conf, pbProtocol, ProtobufRpcEngine.class);
+ RPC.Builder builder = new RPC.Builder(conf)
+ .setProtocol(pbProtocol)
+ .setInstance(blockingService)
+ .setBindAddress(addr.getHostName())
+ .setPort(addr.getPort())
+ .setNumHandlers(numHandlers);
+ if (zkSecretManager != null) {
+ builder = builder.setSecretManager(zkSecretManager);
+ }
+ RPC.Server server = builder.build();
+
+ if (conf.getBoolean(CommonConfigurationKeysPublic.HADOOP_SECURITY_AUTHORIZATION, false)) {
+ server.refreshServiceAcl(conf, new LlapDaemonPolicyProvider());
+ }
+ return server;
+ }
+
+
+ @Override
+ public GetTokenResponseProto getDelegationToken(RpcController controller,
+ GetTokenRequestProto request) throws ServiceException {
+ if (zkSecretManager == null) {
+ throw new ServiceException("Operation not supported on unsecure cluster");
+ }
+ UserGroupInformation ugi;
+ try {
+ ugi = UserGroupInformation.getCurrentUser();
+ } catch (IOException e) {
+ throw new ServiceException(e);
+ }
+ String user = ugi.getUserName();
+ Text owner = new Text(user);
+ Text realUser = null;
+ if (ugi.getRealUser() != null) {
+ realUser = new Text(ugi.getRealUser().getUserName());
+ }
+ Text renewer = new Text(ugi.getShortUserName());
+ LlapTokenIdentifier llapId = new LlapTokenIdentifier(owner, renewer, realUser);
+ // TODO: note that the token is not renewable right now and will last for 2 weeks by default.
+ Token<LlapTokenIdentifier> token = new Token<LlapTokenIdentifier>(llapId, zkSecretManager);
+ ByteArrayDataOutput out = ByteStreams.newDataOutput();
+ try {
+ token.write(out);
+ } catch (IOException e) {
+ throw new ServiceException(e);
+ }
+ ByteString bs = ByteString.copyFrom(out.toByteArray());
+ GetTokenResponseProto response = GetTokenResponseProto.newBuilder().setToken(bs).build();
+ return response;
+ }
+}
http://git-wip-us.apache.org/repos/asf/hive/blob/9886414b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/QueryFragmentInfo.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/QueryFragmentInfo.java b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/QueryFragmentInfo.java
index aa065a9..480a394 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/QueryFragmentInfo.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/QueryFragmentInfo.java
@@ -23,7 +23,7 @@ import org.apache.hadoop.hive.llap.daemon.FinishableStateUpdateHandler;
import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos;
import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.FragmentSpecProto;
import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.IOSpecProto;
-import org.apache.tez.mapreduce.input.MRInputLegacy;
+import org.apache.hadoop.hive.llap.tezplugins.LlapTezUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -89,7 +89,7 @@ public class QueryFragmentInfo {
boolean canFinish = true;
if (inputSpecList != null && !inputSpecList.isEmpty()) {
for (IOSpecProto inputSpec : inputSpecList) {
- if (isSourceOfInterest(inputSpec)) {
+ if (LlapTezUtils.isSourceOfInterest(inputSpec.getIoDescriptor().getClassName())) {
// Lookup the state in the map.
LlapDaemonProtocolProtos.SourceStateProto state = queryInfo.getSourceStateMap()
.get(inputSpec.getConnectedVertexName());
@@ -129,7 +129,7 @@ public class QueryFragmentInfo {
List<IOSpecProto> inputSpecList = fragmentSpec.getInputSpecsList();
if (inputSpecList != null && !inputSpecList.isEmpty()) {
for (IOSpecProto inputSpec : inputSpecList) {
- if (isSourceOfInterest(inputSpec)) {
+ if (LlapTezUtils.isSourceOfInterest(inputSpec.getIoDescriptor().getClassName())) {
sourcesOfInterest.add(inputSpec.getConnectedVertexName());
}
}
@@ -143,13 +143,6 @@ public class QueryFragmentInfo {
queryInfo.unregisterFinishableStateUpdate(handler);
}
- private boolean isSourceOfInterest(IOSpecProto inputSpec) {
- String inputClassName = inputSpec.getIoDescriptor().getClassName();
- // MRInput is not of interest since it'll always be ready.
- return !inputClassName.equals(MRInputLegacy.class.getName());
- }
-
-
@Override
public boolean equals(Object o) {
if (this == o) {
http://git-wip-us.apache.org/repos/asf/hive/blob/9886414b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/TaskRunnerCallable.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/TaskRunnerCallable.java b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/TaskRunnerCallable.java
index ede2a03..d88d82a 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/TaskRunnerCallable.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/TaskRunnerCallable.java
@@ -38,7 +38,7 @@ import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.IOSpecPro
import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.SubmitWorkRequestProto;
import org.apache.hadoop.hive.llap.metrics.LlapDaemonExecutorMetrics;
import org.apache.hadoop.hive.llap.protocol.LlapTaskUmbilicalProtocol;
-import org.apache.hadoop.hive.llap.tezplugins.Converters;
+import org.apache.hadoop.hive.llap.tez.Converters;
import org.apache.hadoop.hive.ql.io.IOContextMap;
import org.apache.hadoop.ipc.RPC;
import org.apache.hadoop.net.NetUtils;
http://git-wip-us.apache.org/repos/asf/hive/blob/9886414b/llap-server/src/java/org/apache/hadoop/hive/llap/protocol/LlapTaskUmbilicalProtocol.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/protocol/LlapTaskUmbilicalProtocol.java b/llap-server/src/java/org/apache/hadoop/hive/llap/protocol/LlapTaskUmbilicalProtocol.java
deleted file mode 100644
index 9549567..0000000
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/protocol/LlapTaskUmbilicalProtocol.java
+++ /dev/null
@@ -1,42 +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.protocol;
-
-import java.io.IOException;
-
-import org.apache.hadoop.io.Text;
-import org.apache.hadoop.ipc.VersionedProtocol;
-import org.apache.hadoop.security.token.TokenInfo;
-import org.apache.tez.dag.api.TezException;
-import org.apache.tez.dag.records.TezTaskAttemptID;
-import org.apache.tez.runtime.api.impl.TezHeartbeatRequest;
-import org.apache.tez.runtime.api.impl.TezHeartbeatResponse;
-import org.apache.tez.runtime.common.security.JobTokenSelector;
-
-@TokenInfo(JobTokenSelector.class)
-public interface LlapTaskUmbilicalProtocol extends VersionedProtocol {
-
- public static final long versionID = 1L;
-
- // From Tez. Eventually changes over to the LLAP protocol and ProtocolBuffers
- boolean canCommit(TezTaskAttemptID taskid) throws IOException;
- public TezHeartbeatResponse heartbeat(TezHeartbeatRequest request)
- throws IOException, TezException;
-
- public void nodeHeartbeat(Text hostname, int port) throws IOException;
-
- public void taskKilled(TezTaskAttemptID taskAttemptId) throws IOException;
-
-}
http://git-wip-us.apache.org/repos/asf/hive/blob/9886414b/llap-server/src/java/org/apache/hadoop/hive/llap/security/LlapDaemonPolicyProvider.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/security/LlapDaemonPolicyProvider.java b/llap-server/src/java/org/apache/hadoop/hive/llap/security/LlapDaemonPolicyProvider.java
index d67647b..bedd265 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/security/LlapDaemonPolicyProvider.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/security/LlapDaemonPolicyProvider.java
@@ -18,17 +18,17 @@
package org.apache.hadoop.hive.llap.security;
import org.apache.hadoop.hive.conf.HiveConf;
-import org.apache.hadoop.hive.llap.daemon.LlapDaemonProtocolBlockingPB;
-import org.apache.hadoop.hive.llap.daemon.LlapManagementProtocolBlockingPB;
+import org.apache.hadoop.hive.llap.protocol.LlapProtocolBlockingPB;
+import org.apache.hadoop.hive.llap.protocol.LlapManagementProtocolPB;
import org.apache.hadoop.security.authorize.PolicyProvider;
import org.apache.hadoop.security.authorize.Service;
public class LlapDaemonPolicyProvider extends PolicyProvider {
private static final Service[] services = new Service[] {
new Service(HiveConf.ConfVars.LLAP_SECURITY_ACL.varname,
- LlapDaemonProtocolBlockingPB.class),
+ LlapProtocolBlockingPB.class),
new Service(HiveConf.ConfVars.LLAP_MANAGEMENT_ACL.varname,
- LlapManagementProtocolBlockingPB.class)
+ LlapManagementProtocolPB.class)
};
@Override
http://git-wip-us.apache.org/repos/asf/hive/blob/9886414b/llap-server/src/java/org/apache/hadoop/hive/llap/security/LlapSecurityHelper.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/security/LlapSecurityHelper.java b/llap-server/src/java/org/apache/hadoop/hive/llap/security/LlapSecurityHelper.java
index a00b631..aa8745d 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/security/LlapSecurityHelper.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/security/LlapSecurityHelper.java
@@ -27,7 +27,7 @@ import java.util.concurrent.TimeUnit;
import javax.net.SocketFactory;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hive.llap.daemon.impl.LlapManagementProtocolClientImpl;
+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;
http://git-wip-us.apache.org/repos/asf/hive/blob/9886414b/llap-server/src/java/org/apache/hadoop/hive/llap/security/LlapServerSecurityInfo.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/security/LlapServerSecurityInfo.java b/llap-server/src/java/org/apache/hadoop/hive/llap/security/LlapServerSecurityInfo.java
index 4dca2ce..eb514f2 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/security/LlapServerSecurityInfo.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/security/LlapServerSecurityInfo.java
@@ -20,8 +20,8 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.conf.HiveConf;
-import org.apache.hadoop.hive.llap.daemon.LlapDaemonProtocolBlockingPB;
-import org.apache.hadoop.hive.llap.daemon.LlapManagementProtocolBlockingPB;
+import org.apache.hadoop.hive.llap.protocol.LlapProtocolBlockingPB;
+import org.apache.hadoop.hive.llap.protocol.LlapManagementProtocolPB;
import org.apache.hadoop.security.KerberosInfo;
import org.apache.hadoop.security.SecurityInfo;
import org.apache.hadoop.security.token.TokenIdentifier;
@@ -36,8 +36,8 @@ public class LlapServerSecurityInfo extends SecurityInfo {
if (LOG.isDebugEnabled()) {
LOG.debug("Trying to get KerberosInfo for " + protocol);
}
- if (!LlapDaemonProtocolBlockingPB.class.isAssignableFrom(protocol)
- && !LlapManagementProtocolBlockingPB.class.isAssignableFrom(protocol)) return null;
+ if (!LlapProtocolBlockingPB.class.isAssignableFrom(protocol)
+ && !LlapManagementProtocolPB.class.isAssignableFrom(protocol)) return null;
return new KerberosInfo() {
@Override
public Class<? extends Annotation> annotationType() {
@@ -62,7 +62,7 @@ public class LlapServerSecurityInfo extends SecurityInfo {
LOG.debug("Trying to get TokenInfo for " + protocol);
}
// Tokens cannot be used for the management protocol (for now).
- if (!LlapDaemonProtocolBlockingPB.class.isAssignableFrom(protocol)) return null;
+ if (!LlapProtocolBlockingPB.class.isAssignableFrom(protocol)) return null;
return new TokenInfo() {
@Override
public Class<? extends Annotation> annotationType() {
http://git-wip-us.apache.org/repos/asf/hive/blob/9886414b/llap-server/src/java/org/apache/hadoop/hive/llap/security/LlapTokenSelector.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/security/LlapTokenSelector.java b/llap-server/src/java/org/apache/hadoop/hive/llap/security/LlapTokenSelector.java
deleted file mode 100644
index b6e7499..0000000
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/security/LlapTokenSelector.java
+++ /dev/null
@@ -1,53 +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.security;
-
-import java.util.Collection;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.hadoop.io.Text;
-import org.apache.hadoop.security.token.Token;
-import org.apache.hadoop.security.token.TokenIdentifier;
-import org.apache.hadoop.security.token.TokenSelector;
-
-public class LlapTokenSelector implements TokenSelector<LlapTokenIdentifier> {
- private static final Log LOG = LogFactory.getLog(LlapTokenSelector.class);
-
- @Override
- public Token<LlapTokenIdentifier> selectToken(Text service,
- Collection<Token<? extends TokenIdentifier>> tokens) {
- if (service == null) return null;
- if (LOG.isDebugEnabled()) {
- LOG.debug("Looking for a token with service " + service);
- }
- for (Token<? extends TokenIdentifier> token : tokens) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Token = " + token.getKind() + "; service = " + token.getService());
- }
- if (LlapTokenIdentifier.KIND_NAME.equals(token.getKind())
- && service.equals(token.getService())) {
- @SuppressWarnings("unchecked")
- Token<LlapTokenIdentifier> result = (Token<LlapTokenIdentifier>)token;
- return result;
- }
- }
- return null;
- }
-}
http://git-wip-us.apache.org/repos/asf/hive/blob/9886414b/llap-server/src/java/org/apache/hadoop/hive/llap/tezplugins/Converters.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/tezplugins/Converters.java b/llap-server/src/java/org/apache/hadoop/hive/llap/tezplugins/Converters.java
deleted file mode 100644
index f61d62f..0000000
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/tezplugins/Converters.java
+++ /dev/null
@@ -1,265 +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.tezplugins;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
-
-import com.google.protobuf.ByteString;
-import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.EntityDescriptorProto;
-import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.GroupInputSpecProto;
-import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.IOSpecProto;
-import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.FragmentSpecProto;
-import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.SourceStateProto;
-import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.UserPayloadProto;
-import org.apache.tez.common.TezCommonUtils;
-import org.apache.tez.dag.api.EntityDescriptor;
-import org.apache.tez.dag.api.InputDescriptor;
-import org.apache.tez.dag.api.OutputDescriptor;
-import org.apache.tez.dag.api.ProcessorDescriptor;
-import org.apache.tez.dag.api.TezUncheckedException;
-import org.apache.tez.dag.api.UserPayload;
-import org.apache.tez.dag.api.event.VertexState;
-import org.apache.tez.dag.records.TezTaskAttemptID;
-import org.apache.tez.runtime.api.impl.GroupInputSpec;
-import org.apache.tez.runtime.api.impl.InputSpec;
-import org.apache.tez.runtime.api.impl.OutputSpec;
-import org.apache.tez.runtime.api.impl.TaskSpec;
-
-public class Converters {
-
- public static TaskSpec getTaskSpecfromProto(FragmentSpecProto FragmentSpecProto) {
- TezTaskAttemptID taskAttemptID =
- TezTaskAttemptID.fromString(FragmentSpecProto.getFragmentIdentifierString());
-
- ProcessorDescriptor processorDescriptor = null;
- if (FragmentSpecProto.hasProcessorDescriptor()) {
- processorDescriptor = convertProcessorDescriptorFromProto(
- FragmentSpecProto.getProcessorDescriptor());
- }
-
- List<InputSpec> inputSpecList = new ArrayList<InputSpec>(FragmentSpecProto.getInputSpecsCount());
- if (FragmentSpecProto.getInputSpecsCount() > 0) {
- for (IOSpecProto inputSpecProto : FragmentSpecProto.getInputSpecsList()) {
- inputSpecList.add(getInputSpecFromProto(inputSpecProto));
- }
- }
-
- List<OutputSpec> outputSpecList =
- new ArrayList<OutputSpec>(FragmentSpecProto.getOutputSpecsCount());
- if (FragmentSpecProto.getOutputSpecsCount() > 0) {
- for (IOSpecProto outputSpecProto : FragmentSpecProto.getOutputSpecsList()) {
- outputSpecList.add(getOutputSpecFromProto(outputSpecProto));
- }
- }
-
- List<GroupInputSpec> groupInputSpecs =
- new ArrayList<GroupInputSpec>(FragmentSpecProto.getGroupedInputSpecsCount());
- if (FragmentSpecProto.getGroupedInputSpecsCount() > 0) {
- for (GroupInputSpecProto groupInputSpecProto : FragmentSpecProto.getGroupedInputSpecsList()) {
- groupInputSpecs.add(getGroupInputSpecFromProto(groupInputSpecProto));
- }
- }
-
- TaskSpec taskSpec =
- new TaskSpec(taskAttemptID, FragmentSpecProto.getDagName(), FragmentSpecProto.getVertexName(),
- FragmentSpecProto.getVertexParallelism(), processorDescriptor, inputSpecList,
- outputSpecList, groupInputSpecs);
- return taskSpec;
- }
-
- public static FragmentSpecProto convertTaskSpecToProto(TaskSpec taskSpec) {
- FragmentSpecProto.Builder builder = FragmentSpecProto.newBuilder();
- builder.setFragmentIdentifierString(taskSpec.getTaskAttemptID().toString());
- builder.setDagName(taskSpec.getDAGName());
- builder.setVertexName(taskSpec.getVertexName());
- builder.setVertexParallelism(taskSpec.getVertexParallelism());
- builder.setFragmentNumber(taskSpec.getTaskAttemptID().getTaskID().getId());
- builder.setAttemptNumber(taskSpec.getTaskAttemptID().getId());
-
- if (taskSpec.getProcessorDescriptor() != null) {
- builder.setProcessorDescriptor(
- convertToProto(taskSpec.getProcessorDescriptor()));
- }
-
- if (taskSpec.getInputs() != null && !taskSpec.getInputs().isEmpty()) {
- for (InputSpec inputSpec : taskSpec.getInputs()) {
- builder.addInputSpecs(convertInputSpecToProto(inputSpec));
- }
- }
-
- if (taskSpec.getOutputs() != null && !taskSpec.getOutputs().isEmpty()) {
- for (OutputSpec outputSpec : taskSpec.getOutputs()) {
- builder.addOutputSpecs(convertOutputSpecToProto(outputSpec));
- }
- }
-
- if (taskSpec.getGroupInputs() != null && !taskSpec.getGroupInputs().isEmpty()) {
- for (GroupInputSpec groupInputSpec : taskSpec.getGroupInputs()) {
- builder.addGroupedInputSpecs(convertGroupInputSpecToProto(groupInputSpec));
-
- }
- }
- return builder.build();
- }
-
- private static ProcessorDescriptor convertProcessorDescriptorFromProto(
- EntityDescriptorProto proto) {
- String className = proto.getClassName();
- UserPayload payload = convertPayloadFromProto(proto);
- ProcessorDescriptor pd = ProcessorDescriptor.create(className);
- setUserPayload(pd, payload);
- return pd;
- }
-
- private static EntityDescriptorProto convertToProto(
- EntityDescriptor<?> descriptor) {
- EntityDescriptorProto.Builder builder = EntityDescriptorProto
- .newBuilder();
- builder.setClassName(descriptor.getClassName());
-
- UserPayload userPayload = descriptor.getUserPayload();
- if (userPayload != null) {
- UserPayloadProto.Builder payloadBuilder = UserPayloadProto.newBuilder();
- if (userPayload.hasPayload()) {
- payloadBuilder.setUserPayload(ByteString.copyFrom(userPayload.getPayload()));
- payloadBuilder.setVersion(userPayload.getVersion());
- }
- builder.setUserPayload(payloadBuilder.build());
- }
- if (descriptor.getHistoryText() != null) {
- try {
- builder.setHistoryText(TezCommonUtils.compressByteArrayToByteString(
- descriptor.getHistoryText().getBytes("UTF-8")));
- } catch (IOException e) {
- throw new TezUncheckedException(e);
- }
- }
- return builder.build();
- }
-
- private static InputSpec getInputSpecFromProto(IOSpecProto inputSpecProto) {
- InputDescriptor inputDescriptor = null;
- if (inputSpecProto.hasIoDescriptor()) {
- inputDescriptor =
- convertInputDescriptorFromProto(inputSpecProto.getIoDescriptor());
- }
- InputSpec inputSpec = new InputSpec(inputSpecProto.getConnectedVertexName(), inputDescriptor,
- inputSpecProto.getPhysicalEdgeCount());
- return inputSpec;
- }
-
- private static InputDescriptor convertInputDescriptorFromProto(
- EntityDescriptorProto proto) {
- String className = proto.getClassName();
- UserPayload payload = convertPayloadFromProto(proto);
- InputDescriptor id = InputDescriptor.create(className);
- setUserPayload(id, payload);
- return id;
- }
-
- private static OutputDescriptor convertOutputDescriptorFromProto(
- EntityDescriptorProto proto) {
- String className = proto.getClassName();
- UserPayload payload = convertPayloadFromProto(proto);
- OutputDescriptor od = OutputDescriptor.create(className);
- setUserPayload(od, payload);
- return od;
- }
-
- private static IOSpecProto convertInputSpecToProto(InputSpec inputSpec) {
- IOSpecProto.Builder builder = IOSpecProto.newBuilder();
- if (inputSpec.getSourceVertexName() != null) {
- builder.setConnectedVertexName(inputSpec.getSourceVertexName());
- }
- if (inputSpec.getInputDescriptor() != null) {
- builder.setIoDescriptor(convertToProto(inputSpec.getInputDescriptor()));
- }
- builder.setPhysicalEdgeCount(inputSpec.getPhysicalEdgeCount());
- return builder.build();
- }
-
- private static OutputSpec getOutputSpecFromProto(IOSpecProto outputSpecProto) {
- OutputDescriptor outputDescriptor = null;
- if (outputSpecProto.hasIoDescriptor()) {
- outputDescriptor =
- convertOutputDescriptorFromProto(outputSpecProto.getIoDescriptor());
- }
- OutputSpec outputSpec =
- new OutputSpec(outputSpecProto.getConnectedVertexName(), outputDescriptor,
- outputSpecProto.getPhysicalEdgeCount());
- return outputSpec;
- }
-
- public static IOSpecProto convertOutputSpecToProto(OutputSpec outputSpec) {
- IOSpecProto.Builder builder = IOSpecProto.newBuilder();
- if (outputSpec.getDestinationVertexName() != null) {
- builder.setConnectedVertexName(outputSpec.getDestinationVertexName());
- }
- if (outputSpec.getOutputDescriptor() != null) {
- builder.setIoDescriptor(convertToProto(outputSpec.getOutputDescriptor()));
- }
- builder.setPhysicalEdgeCount(outputSpec.getPhysicalEdgeCount());
- return builder.build();
- }
-
- private static GroupInputSpec getGroupInputSpecFromProto(GroupInputSpecProto groupInputSpecProto) {
- GroupInputSpec groupSpec = new GroupInputSpec(groupInputSpecProto.getGroupName(),
- groupInputSpecProto.getGroupVerticesList(),
- convertInputDescriptorFromProto(groupInputSpecProto.getMergedInputDescriptor()));
- return groupSpec;
- }
-
- private static GroupInputSpecProto convertGroupInputSpecToProto(GroupInputSpec groupInputSpec) {
- GroupInputSpecProto.Builder builder = GroupInputSpecProto.newBuilder();
- builder.setGroupName(groupInputSpec.getGroupName());
- builder.addAllGroupVertices(groupInputSpec.getGroupVertices());
- builder.setMergedInputDescriptor(convertToProto(groupInputSpec.getMergedInputDescriptor()));
- return builder.build();
- }
-
-
- private static void setUserPayload(EntityDescriptor<?> entity, UserPayload payload) {
- if (payload != null) {
- entity.setUserPayload(payload);
- }
- }
-
- private static UserPayload convertPayloadFromProto(
- EntityDescriptorProto proto) {
- UserPayload userPayload = null;
- if (proto.hasUserPayload()) {
- if (proto.getUserPayload().hasUserPayload()) {
- userPayload =
- UserPayload.create(proto.getUserPayload().getUserPayload().asReadOnlyByteBuffer(), proto.getUserPayload().getVersion());
- } else {
- userPayload = UserPayload.create(null);
- }
- }
- return userPayload;
- }
-
- public static SourceStateProto fromVertexState(VertexState state) {
- switch (state) {
- case SUCCEEDED:
- return SourceStateProto.S_SUCCEEDED;
- case RUNNING:
- return SourceStateProto.S_RUNNING;
- default:
- throw new RuntimeException("Unexpected state: " + state);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/hive/blob/9886414b/llap-server/src/java/org/apache/hadoop/hive/llap/tezplugins/LlapContainerLauncher.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/tezplugins/LlapContainerLauncher.java b/llap-server/src/java/org/apache/hadoop/hive/llap/tezplugins/LlapContainerLauncher.java
deleted file mode 100644
index 07703a2..0000000
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/tezplugins/LlapContainerLauncher.java
+++ /dev/null
@@ -1,43 +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.tezplugins;
-
-import org.apache.tez.serviceplugins.api.ContainerLaunchRequest;
-import org.apache.tez.serviceplugins.api.ContainerLauncher;
-import org.apache.tez.serviceplugins.api.ContainerLauncherContext;
-import org.apache.tez.serviceplugins.api.ContainerStopRequest;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class LlapContainerLauncher extends ContainerLauncher {
- private static final Logger LOG = LoggerFactory.getLogger(LlapContainerLauncher.class);
-
- public LlapContainerLauncher(ContainerLauncherContext containerLauncherContext) {
- super(containerLauncherContext);
- }
-
- @Override
- public void launchContainer(ContainerLaunchRequest containerLaunchRequest) {
- LOG.info("No-op launch for container: " + containerLaunchRequest.getContainerId() +
- " succeeded on host: " + containerLaunchRequest.getNodeId());
- getContext().containerLaunched(containerLaunchRequest.getContainerId());
- }
-
- @Override
- public void stopContainer(ContainerStopRequest containerStopRequest) {
- LOG.info("DEBUG: Ignoring STOP_REQUEST for event: " + containerStopRequest);
- getContext().containerStopRequested(containerStopRequest.getContainerId());
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hive/blob/9886414b/llap-server/src/java/org/apache/hadoop/hive/llap/tezplugins/LlapDaemonProtocolClientProxy.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/tezplugins/LlapDaemonProtocolClientProxy.java b/llap-server/src/java/org/apache/hadoop/hive/llap/tezplugins/LlapDaemonProtocolClientProxy.java
deleted file mode 100644
index 2884e40..0000000
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/tezplugins/LlapDaemonProtocolClientProxy.java
+++ /dev/null
@@ -1,509 +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.tezplugins;
-
-import javax.net.SocketFactory;
-
-import java.io.IOException;
-import java.security.PrivilegedAction;
-import java.util.HashSet;
-import java.util.Iterator;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Set;
-import java.util.concurrent.Callable;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.locks.Condition;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantLock;
-
-import com.google.common.annotations.VisibleForTesting;
-import com.google.common.util.concurrent.FutureCallback;
-import com.google.common.util.concurrent.Futures;
-import com.google.common.util.concurrent.ListenableFuture;
-import com.google.common.util.concurrent.ListeningExecutorService;
-import com.google.common.util.concurrent.MoreExecutors;
-import com.google.common.util.concurrent.ThreadFactoryBuilder;
-import com.google.protobuf.Message;
-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.LlapNodeId;
-import org.apache.hadoop.hive.llap.daemon.LlapDaemonProtocolBlockingPB;
-import org.apache.hadoop.hive.llap.daemon.impl.LlapDaemonProtocolClientImpl;
-import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.QueryCompleteRequestProto;
-import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.QueryCompleteResponseProto;
-import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.SourceStateUpdatedRequestProto;
-import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.SourceStateUpdatedResponseProto;
-import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.SubmitWorkRequestProto;
-import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.SubmitWorkResponseProto;
-import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.TerminateFragmentRequestProto;
-import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.TerminateFragmentResponseProto;
-import org.apache.hadoop.hive.llap.security.LlapTokenIdentifier;
-import org.apache.hadoop.io.retry.RetryPolicies;
-import org.apache.hadoop.io.retry.RetryPolicy;
-import org.apache.hadoop.net.NetUtils;
-import org.apache.hadoop.security.SecurityUtil;
-import org.apache.hadoop.security.UserGroupInformation;
-import org.apache.hadoop.security.token.Token;
-import org.apache.hadoop.service.AbstractService;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class LlapDaemonProtocolClientProxy extends AbstractService {
-
- private static final Logger LOG = LoggerFactory.getLogger(LlapDaemonProtocolClientProxy.class);
-
- private final ConcurrentMap<String, LlapDaemonProtocolBlockingPB> hostProxies;
-
- private final RequestManager requestManager;
- private final RetryPolicy retryPolicy;
- private final SocketFactory socketFactory;
-
- private final ListeningExecutorService requestManagerExecutor;
- private volatile ListenableFuture<Void> requestManagerFuture;
- private final Token<LlapTokenIdentifier> llapToken;
-
- public LlapDaemonProtocolClientProxy(
- int numThreads, Configuration conf, Token<LlapTokenIdentifier> llapToken) {
- super(LlapDaemonProtocolClientProxy.class.getSimpleName());
- this.hostProxies = new ConcurrentHashMap<>();
- this.socketFactory = NetUtils.getDefaultSocketFactory(conf);
- this.llapToken = llapToken;
-
- long connectionTimeout = HiveConf.getTimeVar(conf,
- ConfVars.LLAP_TASK_COMMUNICATOR_CONNECTION_TIMEOUT_MS, TimeUnit.MILLISECONDS);
- long retrySleep = HiveConf.getTimeVar(conf,
- ConfVars.LLAP_TASK_COMMUNICATOR_CONNECTION_SLEEP_BETWEEN_RETRIES_MS,
- TimeUnit.MILLISECONDS);
- this.retryPolicy = RetryPolicies.retryUpToMaximumTimeWithFixedSleep(
- connectionTimeout, retrySleep, TimeUnit.MILLISECONDS);
-
- this.requestManager = new RequestManager(numThreads);
- ExecutorService localExecutor = Executors.newFixedThreadPool(1,
- new ThreadFactoryBuilder().setNameFormat("RequestManagerExecutor").build());
- this.requestManagerExecutor = MoreExecutors.listeningDecorator(localExecutor);
-
- LOG.info("Setting up taskCommunicator with" +
- "numThreads=" + numThreads +
- "retryTime(millis)=" + connectionTimeout +
- "retrySleep(millis)=" + retrySleep);
- }
-
- @Override
- public void serviceStart() {
- requestManagerFuture = requestManagerExecutor.submit(requestManager);
- Futures.addCallback(requestManagerFuture, new FutureCallback<Void>() {
- @Override
- public void onSuccess(Void result) {
- LOG.info("RequestManager shutdown");
- }
-
- @Override
- public void onFailure(Throwable t) {
- LOG.warn("RequestManager shutdown with error", t);
- }
- });
- }
-
- @Override
- public void serviceStop() {
- if (requestManagerFuture != null) {
- requestManager.shutdown();
- requestManagerFuture.cancel(true);
- }
- requestManagerExecutor.shutdown();
- }
-
- public void sendSubmitWork(SubmitWorkRequestProto request, String host, int port,
- final ExecuteRequestCallback<SubmitWorkResponseProto> callback) {
- LlapNodeId nodeId = LlapNodeId.getInstance(host, port);
- requestManager.queueRequest(new SubmitWorkCallable(nodeId, request, callback));
- }
-
- public void sendSourceStateUpdate(final SourceStateUpdatedRequestProto request, final String host,
- final int port,
- final ExecuteRequestCallback<SourceStateUpdatedResponseProto> callback) {
- LlapNodeId nodeId = LlapNodeId.getInstance(host, port);
- requestManager.queueRequest(
- new SendSourceStateUpdateCallable(nodeId, request, callback));
- }
-
- public void sendQueryComplete(final QueryCompleteRequestProto request, final String host,
- final int port,
- final ExecuteRequestCallback<QueryCompleteResponseProto> callback) {
- LlapNodeId nodeId = LlapNodeId.getInstance(host, port);
- requestManager.queueRequest(new SendQueryCompleteCallable(nodeId, request, callback));
- }
-
- public void sendTerminateFragment(final TerminateFragmentRequestProto request, final String host,
- final int port,
- final ExecuteRequestCallback<TerminateFragmentResponseProto> callback) {
- LlapNodeId nodeId = LlapNodeId.getInstance(host, port);
- requestManager.queueRequest(new SendTerminateFragmentCallable(nodeId, request, callback));
- }
-
- @VisibleForTesting
- static class RequestManager implements Callable<Void> {
-
- private final Lock lock = new ReentrantLock();
- private final AtomicBoolean isShutdown = new AtomicBoolean(false);
- private final Condition queueCondition = lock.newCondition();
- private final AtomicBoolean shouldRun = new AtomicBoolean(false);
-
- private final int maxConcurrentRequestsPerNode = 1;
- private final ListeningExecutorService executor;
-
-
- // Tracks new additions via add, while the loop is processing existing ones.
- private final LinkedList<CallableRequest> newRequestList = new LinkedList<>();
-
- // Tracks existing requests which are cycled through.
- private final LinkedList<CallableRequest> pendingRequests = new LinkedList<>();
-
- // Tracks requests executing per node
- private final ConcurrentMap<LlapNodeId, AtomicInteger> runningRequests = new ConcurrentHashMap<>();
-
- // Tracks completed requests pre node
- private final LinkedList<LlapNodeId> completedNodes = new LinkedList<>();
-
- public RequestManager(int numThreads) {
- ExecutorService localExecutor = Executors.newFixedThreadPool(numThreads,
- new ThreadFactoryBuilder().setNameFormat("TaskCommunicator #%2d").build());
- executor = MoreExecutors.listeningDecorator(localExecutor);
- }
-
-
- @VisibleForTesting
- Set<LlapNodeId> currentLoopDisabledNodes = new HashSet<>();
- @VisibleForTesting
- List<CallableRequest> currentLoopSkippedRequests = new LinkedList<>();
- @Override
- public Void call() {
- // Caches disabled nodes for quicker lookups and ensures a request on a node which was skipped
- // does not go out of order.
- while (!isShutdown.get()) {
- lock.lock();
- try {
- while (!shouldRun.get()) {
- queueCondition.await();
- break; // Break out and try executing.
- }
- boolean shouldBreak = process();
- if (shouldBreak) {
- break;
- }
- } catch (InterruptedException e) {
- if (isShutdown.get()) {
- break;
- } else {
- LOG.warn("RunLoop interrupted without being shutdown first");
- throw new RuntimeException(e);
- }
- } finally {
- lock.unlock();
- }
- }
- LOG.info("CallScheduler loop exiting");
- return null;
- }
-
- /* Add a new request to be executed */
- public void queueRequest(CallableRequest request) {
- synchronized (newRequestList) {
- newRequestList.add(request);
- shouldRun.set(true);
- }
- notifyRunLoop();
- }
-
- /* Indicates a request has completed on a node */
- public void requestFinished(LlapNodeId nodeId) {
- synchronized (completedNodes) {
- completedNodes.add(nodeId);
- shouldRun.set(true);
- }
- notifyRunLoop();
- }
-
- public void shutdown() {
- if (!isShutdown.getAndSet(true)) {
- executor.shutdownNow();
- notifyRunLoop();
- }
- }
-
- @VisibleForTesting
- void submitToExecutor(CallableRequest request, LlapNodeId nodeId) {
- ListenableFuture<SourceStateUpdatedResponseProto> future =
- executor.submit(request);
- Futures.addCallback(future, new ResponseCallback(request.getCallback(), nodeId, this));
- }
-
- @VisibleForTesting
- boolean process() {
- if (isShutdown.get()) {
- return true;
- }
- currentLoopDisabledNodes.clear();
- currentLoopSkippedRequests.clear();
-
- // Set to false to block the next loop. This must be called before draining the lists,
- // otherwise an add/completion after draining the lists but before setting it to false,
- // will not trigger a run. May cause one unnecessary run if an add comes in before drain.
- // drain list. add request (setTrue). setFalse needs to be avoided.
- shouldRun.compareAndSet(true, false);
- // Drain any calls which may have come in during the last execution of the loop.
- drainNewRequestList(); // Locks newRequestList
- drainCompletedNodes(); // Locks completedNodes
-
-
- Iterator<CallableRequest> iterator = pendingRequests.iterator();
- while (iterator.hasNext()) {
- CallableRequest request = iterator.next();
- iterator.remove();
- LlapNodeId nodeId = request.getNodeId();
- if (canRunForNode(nodeId, currentLoopDisabledNodes)) {
- submitToExecutor(request, nodeId);
- } else {
- currentLoopDisabledNodes.add(nodeId);
- currentLoopSkippedRequests.add(request);
- }
- }
- // Tried scheduling everything that could be scheduled in this loop.
- pendingRequests.addAll(0, currentLoopSkippedRequests);
- return false;
- }
-
- private void drainNewRequestList() {
- synchronized (newRequestList) {
- if (!newRequestList.isEmpty()) {
- pendingRequests.addAll(newRequestList);
- newRequestList.clear();
- }
- }
- }
-
- private void drainCompletedNodes() {
- synchronized (completedNodes) {
- if (!completedNodes.isEmpty()) {
- for (LlapNodeId nodeId : completedNodes) {
- runningRequests.get(nodeId).decrementAndGet();
- }
- }
- completedNodes.clear();
- }
- }
-
- private boolean canRunForNode(LlapNodeId nodeId, Set<LlapNodeId> currentRunDisabledNodes) {
- if (currentRunDisabledNodes.contains(nodeId)) {
- return false;
- } else {
- AtomicInteger count = runningRequests.get(nodeId);
- if (count == null) {
- count = new AtomicInteger(0);
- AtomicInteger old = runningRequests.putIfAbsent(nodeId, count);
- count = old != null ? old : count;
- }
- if (count.incrementAndGet() <= maxConcurrentRequestsPerNode) {
- return true;
- } else {
- count.decrementAndGet();
- return false;
- }
- }
- }
-
- private void notifyRunLoop() {
- lock.lock();
- try {
- queueCondition.signal();
- } finally {
- lock.unlock();
- }
- }
- }
-
-
- private static final class ResponseCallback<TYPE extends Message>
- implements FutureCallback<TYPE> {
-
- private final ExecuteRequestCallback<TYPE> callback;
- private final LlapNodeId nodeId;
- private final RequestManager requestManager;
-
- public ResponseCallback(ExecuteRequestCallback<TYPE> callback, LlapNodeId nodeId,
- RequestManager requestManager) {
- this.callback = callback;
- this.nodeId = nodeId;
- this.requestManager = requestManager;
- }
-
- @Override
- public void onSuccess(TYPE result) {
- try {
- callback.setResponse(result);
- } finally {
- requestManager.requestFinished(nodeId);
- }
- }
-
- @Override
- public void onFailure(Throwable t) {
- try {
- callback.indicateError(t);
- } finally {
- requestManager.requestFinished(nodeId);
- }
- }
- }
-
- @VisibleForTesting
- static abstract class CallableRequest<REQUEST extends Message, RESPONSE extends Message>
- implements Callable {
-
- final LlapNodeId nodeId;
- final ExecuteRequestCallback<RESPONSE> callback;
- final REQUEST request;
-
-
- protected CallableRequest(LlapNodeId nodeId, REQUEST request, ExecuteRequestCallback<RESPONSE> callback) {
- this.nodeId = nodeId;
- this.request = request;
- this.callback = callback;
- }
-
- public LlapNodeId getNodeId() {
- return nodeId;
- }
-
- public ExecuteRequestCallback<RESPONSE> getCallback() {
- return callback;
- }
-
- public abstract RESPONSE call() throws Exception;
- }
-
- private class SubmitWorkCallable extends CallableRequest<SubmitWorkRequestProto, SubmitWorkResponseProto> {
-
- protected SubmitWorkCallable(LlapNodeId nodeId,
- SubmitWorkRequestProto submitWorkRequestProto,
- ExecuteRequestCallback<SubmitWorkResponseProto> callback) {
- super(nodeId, submitWorkRequestProto, callback);
- }
-
- @Override
- public SubmitWorkResponseProto call() throws Exception {
- return getProxy(nodeId).submitWork(null, request);
- }
- }
-
- private class SendSourceStateUpdateCallable
- extends CallableRequest<SourceStateUpdatedRequestProto, SourceStateUpdatedResponseProto> {
-
- public SendSourceStateUpdateCallable(LlapNodeId nodeId,
- SourceStateUpdatedRequestProto request,
- ExecuteRequestCallback<SourceStateUpdatedResponseProto> callback) {
- super(nodeId, request, callback);
- }
-
- @Override
- public SourceStateUpdatedResponseProto call() throws Exception {
- return getProxy(nodeId).sourceStateUpdated(null, request);
- }
- }
-
- private class SendQueryCompleteCallable
- extends CallableRequest<QueryCompleteRequestProto, QueryCompleteResponseProto> {
-
- protected SendQueryCompleteCallable(LlapNodeId nodeId,
- QueryCompleteRequestProto queryCompleteRequestProto,
- ExecuteRequestCallback<QueryCompleteResponseProto> callback) {
- super(nodeId, queryCompleteRequestProto, callback);
- }
-
- @Override
- public QueryCompleteResponseProto call() throws Exception {
- return getProxy(nodeId).queryComplete(null, request);
- }
- }
-
- private class SendTerminateFragmentCallable
- extends CallableRequest<TerminateFragmentRequestProto, TerminateFragmentResponseProto> {
-
- protected SendTerminateFragmentCallable(LlapNodeId nodeId,
- TerminateFragmentRequestProto terminateFragmentRequestProto,
- ExecuteRequestCallback<TerminateFragmentResponseProto> callback) {
- super(nodeId, terminateFragmentRequestProto, callback);
- }
-
- @Override
- public TerminateFragmentResponseProto call() throws Exception {
- return getProxy(nodeId).terminateFragment(null, request);
- }
- }
-
- public interface ExecuteRequestCallback<T extends Message> {
- void setResponse(T response);
- void indicateError(Throwable t);
- }
-
- private LlapDaemonProtocolBlockingPB getProxy(final LlapNodeId nodeId) {
- String hostId = getHostIdentifier(nodeId.getHostname(), nodeId.getPort());
-
- LlapDaemonProtocolBlockingPB proxy = hostProxies.get(hostId);
- if (proxy == null) {
- if (llapToken == null) {
- proxy = new LlapDaemonProtocolClientImpl(getConfig(), nodeId.getHostname(),
- nodeId.getPort(), retryPolicy, socketFactory);
- } else {
- UserGroupInformation ugi;
- try {
- ugi = UserGroupInformation.getCurrentUser();
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- Token<LlapTokenIdentifier> nodeToken = new Token<LlapTokenIdentifier>(llapToken);
- SecurityUtil.setTokenService(nodeToken, NetUtils.createSocketAddrForHost(
- nodeId.getHostname(), nodeId.getPort()));
- ugi.addToken(nodeToken);
- proxy = ugi.doAs(new PrivilegedAction<LlapDaemonProtocolBlockingPB>() {
- @Override
- public LlapDaemonProtocolBlockingPB run() {
- return new LlapDaemonProtocolClientImpl(getConfig(), nodeId.getHostname(),
- nodeId.getPort(), retryPolicy, socketFactory);
- }
- });
- }
-
- LlapDaemonProtocolBlockingPB proxyOld = hostProxies.putIfAbsent(hostId, proxy);
- if (proxyOld != null) {
- // TODO Shutdown the new proxy.
- proxy = proxyOld;
- }
- }
- return proxy;
- }
-
- private String getHostIdentifier(String hostname, int port) {
- return hostname + ":" + port;
- }
-}