You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by se...@apache.org on 2016/08/08 18:41:48 UTC
hive git commit: HIVE-14422 : LLAP IF: when using LLAP IF from
multiple threads in secure cluster, tokens can get mixed up (Sergey Shelukhin,
reviewed by Siddharth Seth)
Repository: hive
Updated Branches:
refs/heads/master 07c20035f -> 3f16a991b
HIVE-14422 : LLAP IF: when using LLAP IF from multiple threads in secure cluster, tokens can get mixed up (Sergey Shelukhin, reviewed by Siddharth Seth)
Project: http://git-wip-us.apache.org/repos/asf/hive/repo
Commit: http://git-wip-us.apache.org/repos/asf/hive/commit/3f16a991
Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/3f16a991
Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/3f16a991
Branch: refs/heads/master
Commit: 3f16a991ba99a00d95452f06ab308e8abfd191d8
Parents: 07c2003
Author: Sergey Shelukhin <se...@apache.org>
Authored: Mon Aug 8 11:31:07 2016 -0700
Committer: Sergey Shelukhin <se...@apache.org>
Committed: Mon Aug 8 11:31:07 2016 -0700
----------------------------------------------------------------------
.../hive/llap/tez/LlapProtocolClientProxy.java | 22 +++++++++++------
.../hive/llap/impl/LlapProtocolClientImpl.java | 26 +++++++++++++++++++-
.../hive/llap/daemon/impl/LlapTokenChecker.java | 16 ++++++++++++
.../impl/TestLlapDaemonProtocolServerImpl.java | 2 +-
.../hive/llap/security/TestLlapSignerImpl.java | 1 -
.../ql/udf/generic/GenericUDTFGetSplits.java | 1 +
6 files changed, 57 insertions(+), 11 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hive/blob/3f16a991/llap-client/src/java/org/apache/hadoop/hive/llap/tez/LlapProtocolClientProxy.java
----------------------------------------------------------------------
diff --git a/llap-client/src/java/org/apache/hadoop/hive/llap/tez/LlapProtocolClientProxy.java b/llap-client/src/java/org/apache/hadoop/hive/llap/tez/LlapProtocolClientProxy.java
index f48a1cb..ce75d72 100644
--- a/llap-client/src/java/org/apache/hadoop/hive/llap/tez/LlapProtocolClientProxy.java
+++ b/llap-client/src/java/org/apache/hadoop/hive/llap/tez/LlapProtocolClientProxy.java
@@ -81,6 +81,7 @@ public class LlapProtocolClientProxy extends AbstractService {
private final ListeningExecutorService requestManagerExecutor;
private volatile ListenableFuture<Void> requestManagerFuture;
private final Token<LlapTokenIdentifier> llapToken;
+ private final String llapTokenUser;
public LlapProtocolClientProxy(
int numThreads, Configuration conf, Token<LlapTokenIdentifier> llapToken) {
@@ -88,6 +89,15 @@ public class LlapProtocolClientProxy extends AbstractService {
this.hostProxies = new ConcurrentHashMap<>();
this.socketFactory = NetUtils.getDefaultSocketFactory(conf);
this.llapToken = llapToken;
+ if (llapToken != null) {
+ try {
+ llapTokenUser = llapToken.decodeIdentifier().getOwner().toString();
+ } catch (IOException e) {
+ throw new RuntimeException("Cannot determine the user from token " + llapToken, e);
+ }
+ } else {
+ llapTokenUser = null;
+ }
long connectionTimeout = HiveConf.getTimeVar(conf,
ConfVars.LLAP_TASK_COMMUNICATOR_CONNECTION_TIMEOUT_MS, TimeUnit.MILLISECONDS);
@@ -474,14 +484,10 @@ public class LlapProtocolClientProxy extends AbstractService {
LOG.debug("Creating a client without a token for " + nodeId);
}
proxy = new LlapProtocolClientImpl(getConfig(), nodeId.getHostname(),
- nodeId.getPort(), retryPolicy, socketFactory);
+ nodeId.getPort(), null, retryPolicy, socketFactory);
} else {
- UserGroupInformation ugi;
- try {
- ugi = UserGroupInformation.getCurrentUser();
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
+ final UserGroupInformation ugi = UserGroupInformation.createRemoteUser(llapTokenUser);
+ // Clone the token as we'd need to set the service to the one we are talking to.
Token<LlapTokenIdentifier> nodeToken = new Token<LlapTokenIdentifier>(llapToken);
SecurityUtil.setTokenService(nodeToken, NetUtils.createSocketAddrForHost(
nodeId.getHostname(), nodeId.getPort()));
@@ -493,7 +499,7 @@ public class LlapProtocolClientProxy extends AbstractService {
@Override
public LlapProtocolBlockingPB run() {
return new LlapProtocolClientImpl(getConfig(), nodeId.getHostname(),
- nodeId.getPort(), retryPolicy, socketFactory);
+ nodeId.getPort(), ugi, retryPolicy, socketFactory);
}
});
}
http://git-wip-us.apache.org/repos/asf/hive/blob/3f16a991/llap-common/src/java/org/apache/hadoop/hive/llap/impl/LlapProtocolClientImpl.java
----------------------------------------------------------------------
diff --git a/llap-common/src/java/org/apache/hadoop/hive/llap/impl/LlapProtocolClientImpl.java b/llap-common/src/java/org/apache/hadoop/hive/llap/impl/LlapProtocolClientImpl.java
index 9234b7d..3681797 100644
--- a/llap-common/src/java/org/apache/hadoop/hive/llap/impl/LlapProtocolClientImpl.java
+++ b/llap-common/src/java/org/apache/hadoop/hive/llap/impl/LlapProtocolClientImpl.java
@@ -18,6 +18,7 @@ import javax.annotation.Nullable;
import javax.net.SocketFactory;
import java.io.IOException;
import java.net.InetSocketAddress;
+import java.security.PrivilegedExceptionAction;
import com.google.protobuf.RpcController;
import com.google.protobuf.ServiceException;
@@ -37,22 +38,30 @@ import org.apache.hadoop.ipc.RPC;
import org.apache.hadoop.net.NetUtils;
import org.apache.hadoop.hive.llap.protocol.LlapProtocolBlockingPB;
import org.apache.hadoop.security.UserGroupInformation;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
// 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 LlapProtocolClientImpl implements LlapProtocolBlockingPB {
+ private final static Logger LOG = LoggerFactory.getLogger(LlapProtocolClientImpl.class);
private final Configuration conf;
private final InetSocketAddress serverAddr;
private final RetryPolicy retryPolicy;
private final SocketFactory socketFactory;
- LlapProtocolBlockingPB proxy;
+ private LlapProtocolBlockingPB proxy;
+ private final UserGroupInformation ugi;
public LlapProtocolClientImpl(Configuration conf, String hostname, int port,
+ UserGroupInformation ugi,
@Nullable RetryPolicy retryPolicy,
@Nullable SocketFactory socketFactory) {
+ // Technically, methods run on a threadpool that is created externally with the UGI.
+ // However, that is brittle, so we'd save the UGI explicitly here.
+ this.ugi = ugi;
this.conf = conf;
this.serverAddr = NetUtils.createSocketAddr(hostname, port);
this.retryPolicy = retryPolicy;
@@ -116,6 +125,21 @@ public class LlapProtocolClientImpl implements LlapProtocolBlockingPB {
public LlapProtocolBlockingPB createProxy() throws IOException {
RPC.setProtocolEngine(conf, LlapProtocolBlockingPB.class, ProtobufRpcEngine.class);
+ LOG.info("Creating protocol proxy as " + ugi);
+ if (ugi == null) return createProxyInternal();
+ try {
+ return ugi.doAs(new PrivilegedExceptionAction<LlapProtocolBlockingPB>() {
+ @Override
+ public LlapProtocolBlockingPB run() throws IOException {
+ return createProxyInternal();
+ }
+ });
+ } catch (InterruptedException e) {
+ throw new IOException(e);
+ }
+ }
+
+ private LlapProtocolBlockingPB createProxyInternal() throws IOException {
ProtocolProxy<LlapProtocolBlockingPB> proxy =
RPC.getProtocolProxy(LlapProtocolBlockingPB.class, 0, serverAddr,
UserGroupInformation.getCurrentUser(), conf, NetUtils.getDefaultSocketFactory(conf), 0,
http://git-wip-us.apache.org/repos/asf/hive/blob/3f16a991/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/LlapTokenChecker.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/LlapTokenChecker.java b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/LlapTokenChecker.java
index beb5c7f..c606c9b 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/LlapTokenChecker.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/LlapTokenChecker.java
@@ -53,9 +53,25 @@ public final class LlapTokenChecker {
if ((tokens == null || tokens.isEmpty()) && kerberosName == null) {
throw new SecurityException("No tokens or kerberos for " + current);
}
+ warnMultipleTokens(tokens);
return getTokenInfoInternal(kerberosName, tokens);
}
+ public static void warnMultipleTokens(List<LlapTokenIdentifier> tokens) {
+ if (tokens != null && tokens.size() > 1) {
+ StringBuilder sb = new StringBuilder("Found multiple LLAP tokens: [");
+ boolean isFirst = true;
+ for (LlapTokenIdentifier ti : tokens) {
+ if (!isFirst) {
+ sb.append(", ");
+ }
+ isFirst = false;
+ sb.append(ti);
+ }
+ LOG.warn(sb.append("]").toString());
+ }
+ }
+
private static List<LlapTokenIdentifier> getLlapTokens(
UserGroupInformation ugi, String clusterId) {
List<LlapTokenIdentifier> tokens = null;
http://git-wip-us.apache.org/repos/asf/hive/blob/3f16a991/llap-server/src/test/org/apache/hadoop/hive/llap/daemon/impl/TestLlapDaemonProtocolServerImpl.java
----------------------------------------------------------------------
diff --git a/llap-server/src/test/org/apache/hadoop/hive/llap/daemon/impl/TestLlapDaemonProtocolServerImpl.java b/llap-server/src/test/org/apache/hadoop/hive/llap/daemon/impl/TestLlapDaemonProtocolServerImpl.java
index b38e9d6..105a912 100644
--- a/llap-server/src/test/org/apache/hadoop/hive/llap/daemon/impl/TestLlapDaemonProtocolServerImpl.java
+++ b/llap-server/src/test/org/apache/hadoop/hive/llap/daemon/impl/TestLlapDaemonProtocolServerImpl.java
@@ -59,7 +59,7 @@ public class TestLlapDaemonProtocolServerImpl {
LlapProtocolBlockingPB client =
new LlapProtocolClientImpl(new Configuration(), serverAddr.getHostName(),
- serverAddr.getPort(), null, null);
+ serverAddr.getPort(), null, null, null);
SubmitWorkResponseProto responseProto = client.submitWork(null,
SubmitWorkRequestProto.newBuilder()
.setAmHost("amhost")
http://git-wip-us.apache.org/repos/asf/hive/blob/3f16a991/llap-server/src/test/org/apache/hadoop/hive/llap/security/TestLlapSignerImpl.java
----------------------------------------------------------------------
diff --git a/llap-server/src/test/org/apache/hadoop/hive/llap/security/TestLlapSignerImpl.java b/llap-server/src/test/org/apache/hadoop/hive/llap/security/TestLlapSignerImpl.java
index a281fd6..fbce98a 100644
--- a/llap-server/src/test/org/apache/hadoop/hive/llap/security/TestLlapSignerImpl.java
+++ b/llap-server/src/test/org/apache/hadoop/hive/llap/security/TestLlapSignerImpl.java
@@ -169,7 +169,6 @@ public class TestLlapSignerImpl {
implements SigningSecretManager {
public FakeSecretManager() {
- // The keys instantly expire and are rolled.
super(10000000, 10000000, 10000000, 10000000);
}
http://git-wip-us.apache.org/repos/asf/hive/blob/3f16a991/ql/src/java/org/apache/hadoop/hive/ql/udf/generic/GenericUDTFGetSplits.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/udf/generic/GenericUDTFGetSplits.java b/ql/src/java/org/apache/hadoop/hive/ql/udf/generic/GenericUDTFGetSplits.java
index 0705e2e..3741ddf 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/udf/generic/GenericUDTFGetSplits.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/udf/generic/GenericUDTFGetSplits.java
@@ -356,6 +356,7 @@ public class GenericUDTFGetSplits extends GenericUDTF {
// We put the query user, not LLAP user, into the message and token.
Token<LlapTokenIdentifier> token = tokenClient.createToken(
applicationId.toString(), queryUser, true);
+ LOG.info("Created the token for remote user: {}", token);
bos.reset();
token.write(dos);
tokenBytes = bos.toByteArray();