hive-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sp...@apache.org
Subject [03/50] [abbrv] hive git commit: HIVE-13445 : LLAP: token should encode application and cluster ids (Sergey Shelukhin, reviewed by Siddharth Seth)
Date Fri, 06 May 2016 20:42:34 GMT
HIVE-13445 : LLAP: token should encode application and cluster ids (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/868e5e14
Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/868e5e14
Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/868e5e14

Branch: refs/heads/java8
Commit: 868e5e141856ce75af48d854d9e3eb13372d11f4
Parents: b621827
Author: Sergey Shelukhin <sershe@apache.org>
Authored: Tue May 3 12:01:32 2016 -0700
Committer: Sergey Shelukhin <sershe@apache.org>
Committed: Tue May 3 13:38:03 2016 -0700

----------------------------------------------------------------------
 .../daemon/rpc/LlapDaemonProtocolProtos.java    | 209 +++++++++++++++++--
 .../org/apache/hadoop/hive/llap/DaemonId.java   |  41 ++++
 .../hive/llap/security/LlapTokenIdentifier.java |  39 +++-
 .../hive/llap/security/LlapTokenProvider.java   |   2 +-
 .../src/protobuf/LlapDaemonProtocol.proto       |   1 +
 .../hive/llap/daemon/ContainerRunner.java       |   9 +-
 .../llap/daemon/impl/ContainerRunnerImpl.java   |  47 +++--
 .../hive/llap/daemon/impl/LlapDaemon.java       |  52 ++++-
 .../daemon/impl/LlapProtocolServerImpl.java     |  41 ++--
 .../hive/llap/daemon/impl/LlapTokenChecker.java | 137 ++++++++++++
 .../hadoop/hive/llap/daemon/impl/QueryInfo.java |  17 +-
 .../hive/llap/daemon/impl/QueryTracker.java     |  85 +++++---
 .../hadoop/hive/llap/daemon/impl/Scheduler.java |   2 +
 .../llap/daemon/impl/TaskExecutorService.java   |   9 +
 .../hive/llap/security/LlapSecurityHelper.java  |  15 +-
 .../hive/llap/security/SecretManager.java       |  19 +-
 .../hive/llap/daemon/MiniLlapCluster.java       |   2 +-
 .../daemon/impl/TaskExecutorTestHelpers.java    |   2 +-
 .../impl/TestLlapDaemonProtocolServerImpl.java  |   2 +-
 .../llap/daemon/impl/TestLlapTokenChecker.java  |  96 +++++++++
 .../hive/ql/exec/tez/TezSessionState.java       |   3 +-
 21 files changed, 702 insertions(+), 128 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hive/blob/868e5e14/llap-common/src/gen/protobuf/gen-java/org/apache/hadoop/hive/llap/daemon/rpc/LlapDaemonProtocolProtos.java
----------------------------------------------------------------------
diff --git a/llap-common/src/gen/protobuf/gen-java/org/apache/hadoop/hive/llap/daemon/rpc/LlapDaemonProtocolProtos.java b/llap-common/src/gen/protobuf/gen-java/org/apache/hadoop/hive/llap/daemon/rpc/LlapDaemonProtocolProtos.java
index 4ab7b32..820f6be 100644
--- a/llap-common/src/gen/protobuf/gen-java/org/apache/hadoop/hive/llap/daemon/rpc/LlapDaemonProtocolProtos.java
+++ b/llap-common/src/gen/protobuf/gen-java/org/apache/hadoop/hive/llap/daemon/rpc/LlapDaemonProtocolProtos.java
@@ -12821,6 +12821,21 @@ public final class LlapDaemonProtocolProtos {
 
   public interface GetTokenRequestProtoOrBuilder
       extends com.google.protobuf.MessageOrBuilder {
+
+    // optional string app_id = 1;
+    /**
+     * <code>optional string app_id = 1;</code>
+     */
+    boolean hasAppId();
+    /**
+     * <code>optional string app_id = 1;</code>
+     */
+    java.lang.String getAppId();
+    /**
+     * <code>optional string app_id = 1;</code>
+     */
+    com.google.protobuf.ByteString
+        getAppIdBytes();
   }
   /**
    * Protobuf type {@code GetTokenRequestProto}
@@ -12855,6 +12870,7 @@ public final class LlapDaemonProtocolProtos {
         com.google.protobuf.ExtensionRegistryLite extensionRegistry)
         throws com.google.protobuf.InvalidProtocolBufferException {
       initFields();
+      int mutable_bitField0_ = 0;
       com.google.protobuf.UnknownFieldSet.Builder unknownFields =
           com.google.protobuf.UnknownFieldSet.newBuilder();
       try {
@@ -12872,6 +12888,11 @@ public final class LlapDaemonProtocolProtos {
               }
               break;
             }
+            case 10: {
+              bitField0_ |= 0x00000001;
+              appId_ = input.readBytes();
+              break;
+            }
           }
         }
       } catch (com.google.protobuf.InvalidProtocolBufferException e) {
@@ -12911,7 +12932,52 @@ public final class LlapDaemonProtocolProtos {
       return PARSER;
     }
 
+    private int bitField0_;
+    // optional string app_id = 1;
+    public static final int APP_ID_FIELD_NUMBER = 1;
+    private java.lang.Object appId_;
+    /**
+     * <code>optional string app_id = 1;</code>
+     */
+    public boolean hasAppId() {
+      return ((bitField0_ & 0x00000001) == 0x00000001);
+    }
+    /**
+     * <code>optional string app_id = 1;</code>
+     */
+    public java.lang.String getAppId() {
+      java.lang.Object ref = appId_;
+      if (ref instanceof java.lang.String) {
+        return (java.lang.String) ref;
+      } else {
+        com.google.protobuf.ByteString bs = 
+            (com.google.protobuf.ByteString) ref;
+        java.lang.String s = bs.toStringUtf8();
+        if (bs.isValidUtf8()) {
+          appId_ = s;
+        }
+        return s;
+      }
+    }
+    /**
+     * <code>optional string app_id = 1;</code>
+     */
+    public com.google.protobuf.ByteString
+        getAppIdBytes() {
+      java.lang.Object ref = appId_;
+      if (ref instanceof java.lang.String) {
+        com.google.protobuf.ByteString b = 
+            com.google.protobuf.ByteString.copyFromUtf8(
+                (java.lang.String) ref);
+        appId_ = b;
+        return b;
+      } else {
+        return (com.google.protobuf.ByteString) ref;
+      }
+    }
+
     private void initFields() {
+      appId_ = "";
     }
     private byte memoizedIsInitialized = -1;
     public final boolean isInitialized() {
@@ -12925,6 +12991,9 @@ public final class LlapDaemonProtocolProtos {
     public void writeTo(com.google.protobuf.CodedOutputStream output)
                         throws java.io.IOException {
       getSerializedSize();
+      if (((bitField0_ & 0x00000001) == 0x00000001)) {
+        output.writeBytes(1, getAppIdBytes());
+      }
       getUnknownFields().writeTo(output);
     }
 
@@ -12934,6 +13003,10 @@ public final class LlapDaemonProtocolProtos {
       if (size != -1) return size;
 
       size = 0;
+      if (((bitField0_ & 0x00000001) == 0x00000001)) {
+        size += com.google.protobuf.CodedOutputStream
+          .computeBytesSize(1, getAppIdBytes());
+      }
       size += getUnknownFields().getSerializedSize();
       memoizedSerializedSize = size;
       return size;
@@ -12957,6 +13030,11 @@ public final class LlapDaemonProtocolProtos {
       org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.GetTokenRequestProto other = (org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.GetTokenRequestProto) obj;
 
       boolean result = true;
+      result = result && (hasAppId() == other.hasAppId());
+      if (hasAppId()) {
+        result = result && getAppId()
+            .equals(other.getAppId());
+      }
       result = result &&
           getUnknownFields().equals(other.getUnknownFields());
       return result;
@@ -12970,6 +13048,10 @@ public final class LlapDaemonProtocolProtos {
       }
       int hash = 41;
       hash = (19 * hash) + getDescriptorForType().hashCode();
+      if (hasAppId()) {
+        hash = (37 * hash) + APP_ID_FIELD_NUMBER;
+        hash = (53 * hash) + getAppId().hashCode();
+      }
       hash = (29 * hash) + getUnknownFields().hashCode();
       memoizedHashCode = hash;
       return hash;
@@ -13079,6 +13161,8 @@ public final class LlapDaemonProtocolProtos {
 
       public Builder clear() {
         super.clear();
+        appId_ = "";
+        bitField0_ = (bitField0_ & ~0x00000001);
         return this;
       }
 
@@ -13105,6 +13189,13 @@ public final class LlapDaemonProtocolProtos {
 
       public org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.GetTokenRequestProto buildPartial() {
         org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.GetTokenRequestProto result = new org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.GetTokenRequestProto(this);
+        int from_bitField0_ = bitField0_;
+        int to_bitField0_ = 0;
+        if (((from_bitField0_ & 0x00000001) == 0x00000001)) {
+          to_bitField0_ |= 0x00000001;
+        }
+        result.appId_ = appId_;
+        result.bitField0_ = to_bitField0_;
         onBuilt();
         return result;
       }
@@ -13120,6 +13211,11 @@ public final class LlapDaemonProtocolProtos {
 
       public Builder mergeFrom(org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.GetTokenRequestProto other) {
         if (other == org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.GetTokenRequestProto.getDefaultInstance()) return this;
+        if (other.hasAppId()) {
+          bitField0_ |= 0x00000001;
+          appId_ = other.appId_;
+          onChanged();
+        }
         this.mergeUnknownFields(other.getUnknownFields());
         return this;
       }
@@ -13145,6 +13241,81 @@ public final class LlapDaemonProtocolProtos {
         }
         return this;
       }
+      private int bitField0_;
+
+      // optional string app_id = 1;
+      private java.lang.Object appId_ = "";
+      /**
+       * <code>optional string app_id = 1;</code>
+       */
+      public boolean hasAppId() {
+        return ((bitField0_ & 0x00000001) == 0x00000001);
+      }
+      /**
+       * <code>optional string app_id = 1;</code>
+       */
+      public java.lang.String getAppId() {
+        java.lang.Object ref = appId_;
+        if (!(ref instanceof java.lang.String)) {
+          java.lang.String s = ((com.google.protobuf.ByteString) ref)
+              .toStringUtf8();
+          appId_ = s;
+          return s;
+        } else {
+          return (java.lang.String) ref;
+        }
+      }
+      /**
+       * <code>optional string app_id = 1;</code>
+       */
+      public com.google.protobuf.ByteString
+          getAppIdBytes() {
+        java.lang.Object ref = appId_;
+        if (ref instanceof String) {
+          com.google.protobuf.ByteString b = 
+              com.google.protobuf.ByteString.copyFromUtf8(
+                  (java.lang.String) ref);
+          appId_ = b;
+          return b;
+        } else {
+          return (com.google.protobuf.ByteString) ref;
+        }
+      }
+      /**
+       * <code>optional string app_id = 1;</code>
+       */
+      public Builder setAppId(
+          java.lang.String value) {
+        if (value == null) {
+    throw new NullPointerException();
+  }
+  bitField0_ |= 0x00000001;
+        appId_ = value;
+        onChanged();
+        return this;
+      }
+      /**
+       * <code>optional string app_id = 1;</code>
+       */
+      public Builder clearAppId() {
+        bitField0_ = (bitField0_ & ~0x00000001);
+        appId_ = getDefaultInstance().getAppId();
+        onChanged();
+        return this;
+      }
+      /**
+       * <code>optional string app_id = 1;</code>
+       */
+      public Builder setAppIdBytes(
+          com.google.protobuf.ByteString value) {
+        if (value == null) {
+    throw new NullPointerException();
+  }
+  bitField0_ |= 0x00000001;
+        appId_ = value;
+        onChanged();
+        return this;
+      }
 
       // @@protoc_insertion_point(builder_scope:GetTokenRequestProto)
     }
@@ -14414,24 +14585,24 @@ public final class LlapDaemonProtocolProtos {
       "RequestProto\022/\n\020query_identifier\030\001 \001(\0132\025" +
       ".QueryIdentifierProto\022\"\n\032fragment_identi" +
       "fier_string\030\002 \001(\t\" \n\036TerminateFragmentRe" +
-      "sponseProto\"\026\n\024GetTokenRequestProto\"&\n\025G",
-      "etTokenResponseProto\022\r\n\005token\030\001 \001(\014*2\n\020S" +
-      "ourceStateProto\022\017\n\013S_SUCCEEDED\020\001\022\r\n\tS_RU" +
-      "NNING\020\002*E\n\024SubmissionStateProto\022\014\n\010ACCEP" +
-      "TED\020\001\022\014\n\010REJECTED\020\002\022\021\n\rEVICTED_OTHER\020\0032\316" +
-      "\002\n\022LlapDaemonProtocol\022?\n\nsubmitWork\022\027.Su" +
-      "bmitWorkRequestProto\032\030.SubmitWorkRespons" +
-      "eProto\022W\n\022sourceStateUpdated\022\037.SourceSta" +
-      "teUpdatedRequestProto\032 .SourceStateUpdat" +
-      "edResponseProto\022H\n\rqueryComplete\022\032.Query" +
-      "CompleteRequestProto\032\033.QueryCompleteResp",
-      "onseProto\022T\n\021terminateFragment\022\036.Termina" +
-      "teFragmentRequestProto\032\037.TerminateFragme" +
-      "ntResponseProto2]\n\026LlapManagementProtoco" +
-      "l\022C\n\022getDelegationToken\022\025.GetTokenReques" +
-      "tProto\032\026.GetTokenResponseProtoBH\n&org.ap" +
-      "ache.hadoop.hive.llap.daemon.rpcB\030LlapDa" +
-      "emonProtocolProtos\210\001\001\240\001\001"
+      "sponseProto\"&\n\024GetTokenRequestProto\022\016\n\006a",
+      "pp_id\030\001 \001(\t\"&\n\025GetTokenResponseProto\022\r\n\005" +
+      "token\030\001 \001(\014*2\n\020SourceStateProto\022\017\n\013S_SUC" +
+      "CEEDED\020\001\022\r\n\tS_RUNNING\020\002*E\n\024SubmissionSta" +
+      "teProto\022\014\n\010ACCEPTED\020\001\022\014\n\010REJECTED\020\002\022\021\n\rE" +
+      "VICTED_OTHER\020\0032\316\002\n\022LlapDaemonProtocol\022?\n" +
+      "\nsubmitWork\022\027.SubmitWorkRequestProto\032\030.S" +
+      "ubmitWorkResponseProto\022W\n\022sourceStateUpd" +
+      "ated\022\037.SourceStateUpdatedRequestProto\032 ." +
+      "SourceStateUpdatedResponseProto\022H\n\rquery" +
+      "Complete\022\032.QueryCompleteRequestProto\032\033.Q",
+      "ueryCompleteResponseProto\022T\n\021terminateFr" +
+      "agment\022\036.TerminateFragmentRequestProto\032\037" +
+      ".TerminateFragmentResponseProto2]\n\026LlapM" +
+      "anagementProtocol\022C\n\022getDelegationToken\022" +
+      "\025.GetTokenRequestProto\032\026.GetTokenRespons" +
+      "eProtoBH\n&org.apache.hadoop.hive.llap.da" +
+      "emon.rpcB\030LlapDaemonProtocolProtos\210\001\001\240\001\001"
     };
     com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner assigner =
       new com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner() {
@@ -14533,7 +14704,7 @@ public final class LlapDaemonProtocolProtos {
           internal_static_GetTokenRequestProto_fieldAccessorTable = new
             com.google.protobuf.GeneratedMessage.FieldAccessorTable(
               internal_static_GetTokenRequestProto_descriptor,
-              new java.lang.String[] { });
+              new java.lang.String[] { "AppId", });
           internal_static_GetTokenResponseProto_descriptor =
             getDescriptor().getMessageTypes().get(16);
           internal_static_GetTokenResponseProto_fieldAccessorTable = new

http://git-wip-us.apache.org/repos/asf/hive/blob/868e5e14/llap-common/src/java/org/apache/hadoop/hive/llap/DaemonId.java
----------------------------------------------------------------------
diff --git a/llap-common/src/java/org/apache/hadoop/hive/llap/DaemonId.java b/llap-common/src/java/org/apache/hadoop/hive/llap/DaemonId.java
new file mode 100644
index 0000000..18355e6
--- /dev/null
+++ b/llap-common/src/java/org/apache/hadoop/hive/llap/DaemonId.java
@@ -0,0 +1,41 @@
+/*
+ * 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;
+
+public class DaemonId {
+  private final String userName;
+  private final String clusterName;
+  private final String appId;
+  private final String hostName;
+  private final long startTime;
+
+  public DaemonId(String userName, String clusterName, String hostName, String appId,
+      long startTime) {
+    this.userName = userName;
+    this.clusterName = clusterName;
+    this.appId = appId;
+    this.hostName = hostName;
+    this.startTime = startTime;
+    // TODO: we could also get an unique number per daemon.
+  }
+
+  public String getClusterString() {
+    return userName + "_" + clusterName + "_" + appId;
+  }
+
+  public String getApplicationId() {
+    return appId;
+  }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/hive/blob/868e5e14/llap-common/src/java/org/apache/hadoop/hive/llap/security/LlapTokenIdentifier.java
----------------------------------------------------------------------
diff --git a/llap-common/src/java/org/apache/hadoop/hive/llap/security/LlapTokenIdentifier.java b/llap-common/src/java/org/apache/hadoop/hive/llap/security/LlapTokenIdentifier.java
index 23980d0..e28eddd 100644
--- a/llap-common/src/java/org/apache/hadoop/hive/llap/security/LlapTokenIdentifier.java
+++ b/llap-common/src/java/org/apache/hadoop/hive/llap/security/LlapTokenIdentifier.java
@@ -22,6 +22,7 @@ import java.io.DataInput;
 import java.io.DataOutput;
 import java.io.IOException;
 
+import org.apache.commons.lang.StringUtils;
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.security.token.Token;
@@ -31,25 +32,32 @@ import org.apache.hadoop.security.token.delegation.AbstractDelegationTokenIdenti
 public class LlapTokenIdentifier extends AbstractDelegationTokenIdentifier {
   private static final String KIND = "LLAP_TOKEN";
   public static final Text KIND_NAME = new Text(KIND);
+  private String clusterId;
+  private String appId;
 
   public LlapTokenIdentifier() {
     super();
   }
 
-  public LlapTokenIdentifier(Text owner, Text renewer, Text realUser) {
+  public LlapTokenIdentifier(Text owner, Text renewer, Text realUser,
+      String clusterId, String appId) {
     super(owner, renewer, realUser);
+    this.clusterId = clusterId;
+    this.appId = appId == null ? "" : appId;
   }
 
   @Override
   public void write(DataOutput out) throws IOException {
     super.write(out);
-    // Nothing right now.
+    out.writeUTF(clusterId);
+    out.writeUTF(appId);
   }
 
   @Override
   public void readFields(DataInput in) throws IOException {
     super.readFields(in);
-    // Nothing right now.
+    clusterId = in.readUTF();
+    appId = in.readUTF();
   }
 
   @Override
@@ -57,21 +65,34 @@ public class LlapTokenIdentifier extends AbstractDelegationTokenIdentifier {
     return KIND_NAME;
   }
 
+  public String getAppId() {
+    return appId;
+  }
+
+  public String getClusterId() {
+    return clusterId;
+  }
+
   @Override
   public int hashCode() {
-    // Nothing else right now.
-    return super.hashCode();
+    final int prime = 31;
+    int result = prime * super.hashCode() + ((appId == null) ? 0 : appId.hashCode());
+    return prime * result + ((clusterId == null) ? 0 : clusterId.hashCode());
   }
 
   @Override
-  public boolean equals(Object other) {
-    // Nothing else right now.
-    return super.equals(other);
+  public boolean equals(Object obj) {
+    if (this == obj) return true;
+    if (!(obj instanceof LlapTokenIdentifier) || !super.equals(obj)) return false;
+    LlapTokenIdentifier other = (LlapTokenIdentifier) obj;
+    return (appId == null ? other.appId == null : appId.equals(other.appId))
+        && (clusterId == null ? other.clusterId == null : clusterId.equals(other.clusterId));
   }
 
   @Override
   public String toString() {
-    return KIND + "; " + super.toString();
+    return KIND + "; " + super.toString() + ", cluster " + clusterId + ", app secret hash "
+        + (StringUtils.isBlank(appId) ? 0 : appId.hashCode());
   }
 
   @InterfaceAudience.Private

http://git-wip-us.apache.org/repos/asf/hive/blob/868e5e14/llap-common/src/java/org/apache/hadoop/hive/llap/security/LlapTokenProvider.java
----------------------------------------------------------------------
diff --git a/llap-common/src/java/org/apache/hadoop/hive/llap/security/LlapTokenProvider.java b/llap-common/src/java/org/apache/hadoop/hive/llap/security/LlapTokenProvider.java
index 2e99a28..edf9b18 100644
--- a/llap-common/src/java/org/apache/hadoop/hive/llap/security/LlapTokenProvider.java
+++ b/llap-common/src/java/org/apache/hadoop/hive/llap/security/LlapTokenProvider.java
@@ -23,5 +23,5 @@ import java.io.IOException;
 import org.apache.hadoop.security.token.Token;
 
 public interface LlapTokenProvider {
-  Token<LlapTokenIdentifier> getDelegationToken() throws IOException;
+  Token<LlapTokenIdentifier> getDelegationToken(String appId) throws IOException;
 }
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/hive/blob/868e5e14/llap-common/src/protobuf/LlapDaemonProtocol.proto
----------------------------------------------------------------------
diff --git a/llap-common/src/protobuf/LlapDaemonProtocol.proto b/llap-common/src/protobuf/LlapDaemonProtocol.proto
index 944c96c..5cdc02e 100644
--- a/llap-common/src/protobuf/LlapDaemonProtocol.proto
+++ b/llap-common/src/protobuf/LlapDaemonProtocol.proto
@@ -130,6 +130,7 @@ message TerminateFragmentResponseProto {
 }
 
 message GetTokenRequestProto {
+  optional string app_id = 1;
 }
 
 message GetTokenResponseProto {

http://git-wip-us.apache.org/repos/asf/hive/blob/868e5e14/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/ContainerRunner.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/ContainerRunner.java b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/ContainerRunner.java
index fc29371..c346aed 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/ContainerRunner.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/ContainerRunner.java
@@ -29,9 +29,12 @@ public interface ContainerRunner {
 
   SubmitWorkResponseProto submitWork(SubmitWorkRequestProto request) throws IOException;
 
-  SourceStateUpdatedResponseProto sourceStateUpdated(SourceStateUpdatedRequestProto request);
+  SourceStateUpdatedResponseProto sourceStateUpdated(
+      SourceStateUpdatedRequestProto request) throws IOException;
 
-  QueryCompleteResponseProto queryComplete(QueryCompleteRequestProto request);
+  QueryCompleteResponseProto queryComplete(
+      QueryCompleteRequestProto request) throws IOException;
 
-  TerminateFragmentResponseProto terminateFragment(TerminateFragmentRequestProto request);
+  TerminateFragmentResponseProto terminateFragment(
+      TerminateFragmentRequestProto request)  throws IOException;
 }

http://git-wip-us.apache.org/repos/asf/hive/blob/868e5e14/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/ContainerRunnerImpl.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/ContainerRunnerImpl.java b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/ContainerRunnerImpl.java
index 3d45c7a..78b37f7 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/ContainerRunnerImpl.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/ContainerRunnerImpl.java
@@ -92,7 +92,7 @@ public class ContainerRunnerImpl extends CompositeService implements ContainerRu
       boolean enablePreemption, String[] localDirsBase, AtomicReference<Integer> localShufflePort,
       AtomicReference<InetSocketAddress> localAddress,
       long totalMemoryAvailableBytes, LlapDaemonExecutorMetrics metrics,
-      AMReporter amReporter, ClassLoader classLoader) {
+      AMReporter amReporter, ClassLoader classLoader, String clusterId) {
     super("ContainerRunnerImpl");
     this.conf = conf;
     Preconditions.checkState(numExecutors > 0,
@@ -101,7 +101,7 @@ public class ContainerRunnerImpl extends CompositeService implements ContainerRu
     this.localShufflePort = localShufflePort;
     this.amReporter = amReporter;
 
-    this.queryTracker = new QueryTracker(conf, localDirsBase);
+    this.queryTracker = new QueryTracker(conf, localDirsBase, clusterId);
     addIfService(queryTracker);
     String waitQueueSchedulerClassName = HiveConf.getVar(
         conf, ConfVars.LLAP_DAEMON_WAIT_QUEUE_COMPARATOR_CLASS_NAME);
@@ -175,7 +175,8 @@ public class ContainerRunnerImpl extends CompositeService implements ContainerRu
           fragmentSpec.getFragmentIdentifierString());
       int dagIdentifier = taskAttemptId.getTaskID().getVertexID().getDAGId().getId();
 
-      QueryIdentifier queryIdentifier = new QueryIdentifier(request.getApplicationIdString(), dagIdentifier);
+      QueryIdentifier queryIdentifier = new QueryIdentifier(
+          request.getApplicationIdString(), dagIdentifier);
 
       Credentials credentials = new Credentials();
       DataInputBuffer dib = new DataInputBuffer();
@@ -193,6 +194,7 @@ public class ContainerRunnerImpl extends CompositeService implements ContainerRu
               fragmentSpec.getAttemptNumber(), request.getUser(), request.getFragmentSpec(),
               jobToken);
 
+
       String[] localDirs = fragmentInfo.getLocalDirs();
       Preconditions.checkNotNull(localDirs);
       if (LOG.isDebugEnabled()) {
@@ -200,7 +202,6 @@ public class ContainerRunnerImpl extends CompositeService implements ContainerRu
       }
       // May need to setup localDir for re-localization, which is usually setup as Environment.PWD.
       // Used for re-localization, to add the user specified configuration (conf_pb_binary_stream)
-
       TaskRunnerCallable callable = new TaskRunnerCallable(request, fragmentInfo, new Configuration(getConfig()),
           new LlapExecutionContext(localAddress.get().getHostName(), queryTracker), env,
           credentials, memoryPerExecutor, amReporter, confParams, metrics, killedTaskHandler,
@@ -248,24 +249,23 @@ public class ContainerRunnerImpl extends CompositeService implements ContainerRu
 
   @Override
   public SourceStateUpdatedResponseProto sourceStateUpdated(
-      SourceStateUpdatedRequestProto request) {
+      SourceStateUpdatedRequestProto request) throws IOException {
     LOG.info("Processing state update: " + stringifySourceStateUpdateRequest(request));
-    queryTracker.registerSourceStateChange(
-        new QueryIdentifier(request.getQueryIdentifier().getAppIdentifier(),
-            request.getQueryIdentifier().getDagIdentifier()), request.getSrcName(),
-        request.getState());
+    QueryIdentifier queryId = new QueryIdentifier(request.getQueryIdentifier().getAppIdentifier(),
+        request.getQueryIdentifier().getDagIdentifier());
+    queryTracker.registerSourceStateChange(queryId, request.getSrcName(), request.getState());
     return SourceStateUpdatedResponseProto.getDefaultInstance();
   }
 
   @Override
-  public QueryCompleteResponseProto queryComplete(QueryCompleteRequestProto request) {
+  public QueryCompleteResponseProto queryComplete(
+      QueryCompleteRequestProto request) throws IOException {
     QueryIdentifier queryIdentifier =
         new QueryIdentifier(request.getQueryIdentifier().getAppIdentifier(),
             request.getQueryIdentifier().getDagIdentifier());
     LOG.info("Processing queryComplete notification for {}", queryIdentifier);
-    List<QueryFragmentInfo> knownFragments =
-        queryTracker
-            .queryComplete(queryIdentifier, request.getDeleteDelay());
+    List<QueryFragmentInfo> knownFragments = queryTracker.queryComplete(
+        queryIdentifier, request.getDeleteDelay(), false);
     LOG.info("DBG: Pending fragment count for completed query {} = {}", queryIdentifier,
         knownFragments.size());
     for (QueryFragmentInfo fragmentInfo : knownFragments) {
@@ -277,9 +277,16 @@ public class ContainerRunnerImpl extends CompositeService implements ContainerRu
   }
 
   @Override
-  public TerminateFragmentResponseProto terminateFragment(TerminateFragmentRequestProto request) {
-    LOG.info("DBG: Received terminateFragment request for {}", request.getFragmentIdentifierString());
-    executorService.killFragment(request.getFragmentIdentifierString());
+  public TerminateFragmentResponseProto terminateFragment(
+      TerminateFragmentRequestProto request) throws IOException {
+    String fragmentId = request.getFragmentIdentifierString();
+    LOG.info("DBG: Received terminateFragment request for {}", fragmentId);
+    // TODO: ideally, QueryTracker should have fragment-to-query mapping.
+    QueryIdentifier queryId = executorService.findQueryByFragment(fragmentId);
+    // checkPermissions returns false if query is not found, throws on failure.
+    if (queryId != null && queryTracker.checkPermissionsForQuery(queryId)) {
+      executorService.killFragment(fragmentId);
+    }
     return TerminateFragmentResponseProto.getDefaultInstance();
   }
 
@@ -355,8 +362,12 @@ public class ContainerRunnerImpl extends CompositeService implements ContainerRu
   @Override
   public void queryFailed(QueryIdentifier queryIdentifier) {
     LOG.info("Processing query failed notification for {}", queryIdentifier);
-    List<QueryFragmentInfo> knownFragments =
-        queryTracker.queryComplete(queryIdentifier, -1);
+    List<QueryFragmentInfo> knownFragments;
+    try {
+      knownFragments = queryTracker.queryComplete(queryIdentifier, -1, true);
+    } catch (IOException e) {
+      throw new RuntimeException(e); // Should never happen here, no permission check.
+    }
     LOG.info("DBG: Pending fragment count for failed query {} = {}", queryIdentifier,
         knownFragments.size());
     for (QueryFragmentInfo fragmentInfo : knownFragments) {

http://git-wip-us.apache.org/repos/asf/hive/blob/868e5e14/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 63cb16b..d23a44a 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
@@ -26,12 +26,14 @@ import java.util.List;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.regex.Pattern;
 
 import javax.management.ObjectName;
 
 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.DaemonId;
 import org.apache.hadoop.hive.llap.LlapUtil;
 import org.apache.hadoop.hive.llap.configuration.LlapDaemonConfiguration;
 import org.apache.hadoop.hive.llap.daemon.ContainerRunner;
@@ -57,11 +59,13 @@ import org.apache.hadoop.hive.ql.udf.generic.GenericUDF;
 import org.apache.hadoop.hive.ql.udf.generic.GenericUDFBridge;
 import org.apache.hadoop.hive.ql.udf.generic.GenericUDFBridge.UdfWhitelistChecker;
 import org.apache.hadoop.metrics2.util.MBeans;
+import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.service.CompositeService;
 import org.apache.hadoop.util.ExitUtil;
 import org.apache.hadoop.util.JvmPauseMonitor;
 import org.apache.hadoop.util.StringUtils;
 import org.apache.hadoop.yarn.api.ApplicationConstants;
+import org.apache.hadoop.yarn.util.ConverterUtils;
 import org.apache.hive.common.util.ShutdownHookManager;
 import org.apache.logging.log4j.core.config.Configurator;
 import org.slf4j.Logger;
@@ -97,6 +101,13 @@ public class LlapDaemon extends CompositeService implements ContainerRunner, Lla
   private final int numExecutors;
   private final long maxJvmMemory;
   private final String[] localDirs;
+  private final DaemonId daemonId;
+
+  private final static Pattern hostsRe = Pattern.compile("[^A-Za-z0-9_-]");
+  private static String generateClusterName(Configuration conf) {
+    String hosts = HiveConf.getTrimmedVar(conf, ConfVars.LLAP_DAEMON_SERVICE_HOSTS);
+    return hostsRe.matcher(hosts.startsWith("@") ? hosts.substring(1) : hosts).replaceAll("_");
+  }
 
   // TODO Not the best way to share the address
   private final AtomicReference<InetSocketAddress> srvAddress = new AtomicReference<>(),
@@ -105,11 +116,10 @@ public class LlapDaemon extends CompositeService implements ContainerRunner, Lla
 
   public LlapDaemon(Configuration daemonConf, int numExecutors, long executorMemoryBytes,
       boolean ioEnabled, boolean isDirectCache, long ioMemoryBytes, String[] localDirs, int srvPort,
-      int mngPort, int shufflePort, int webPort) {
+      int mngPort, int shufflePort, int webPort, String appName) {
     super("LlapDaemon");
 
     initializeLogging();
-
     printAsciiArt();
 
     Preconditions.checkArgument(numExecutors > 0);
@@ -129,6 +139,14 @@ public class LlapDaemon extends CompositeService implements ContainerRunner, Lla
           "LLAP service hosts startswith '@' but hive.zookeeper.quorum is not set." +
               " hive.zookeeper.quorum must be set.");
     }
+    String hostName = MetricsUtils.getHostName();
+    try {
+      daemonId = new DaemonId(UserGroupInformation.getCurrentUser().getUserName(),
+          generateClusterName(daemonConf), hostName, appName, System.currentTimeMillis());
+    } catch (IOException ex) {
+      throw new RuntimeException(ex);
+    }
+
 
     this.maxJvmMemory = getTotalHeapSize();
     this.llapIoEnabled = ioEnabled;
@@ -193,7 +211,7 @@ public class LlapDaemon extends CompositeService implements ContainerRunner, Lla
     LlapMetricsSystem.initialize("LlapDaemon");
     this.pauseMonitor = new JvmPauseMonitor(daemonConf);
     pauseMonitor.start();
-    String displayName = "LlapDaemonExecutorMetrics-" + MetricsUtils.getHostName();
+    String displayName = "LlapDaemonExecutorMetrics-" + hostName;
     String sessionId = MetricsUtils.getUUID();
     daemonConf.set("llap.daemon.metrics.sessionid", sessionId);
     String[] strIntervals = HiveConf.getTrimmedStringsVar(daemonConf,
@@ -223,11 +241,11 @@ public class LlapDaemon extends CompositeService implements ContainerRunner, Lla
     this.amReporter = new AMReporter(srvAddress, new QueryFailedHandlerProxy(), daemonConf);
 
     this.server = new LlapProtocolServerImpl(
-        numHandlers, this, srvAddress, mngAddress, srvPort, mngPort);
+        numHandlers, this, srvAddress, mngAddress, srvPort, mngPort, daemonId);
 
     this.containerRunner = new ContainerRunnerImpl(daemonConf, numExecutors, waitQueueSize,
         enablePreemption, localDirs, this.shufflePort, srvAddress, executorMemoryBytes, metrics,
-        amReporter, executorClassLoader);
+        amReporter, executorClassLoader, daemonId.getClusterString());
     addIfService(containerRunner);
 
     // Not adding the registry as a service, since we need to control when it is initialized - conf used to pickup properties.
@@ -377,10 +395,18 @@ public class LlapDaemon extends CompositeService implements ContainerRunner, Lla
       LlapDaemonConfiguration daemonConf = new LlapDaemonConfiguration();
 
       String containerIdStr = System.getenv(ApplicationConstants.Environment.CONTAINER_ID.name());
+      String appName = null;
       if (containerIdStr != null && !containerIdStr.isEmpty()) {
         daemonConf.set(ConfVars.LLAP_DAEMON_CONTAINER_ID.varname, containerIdStr);
+        appName = ConverterUtils.toContainerId(containerIdStr)
+            .getApplicationAttemptId().getApplicationId().toString();
       } else {
         daemonConf.unset(ConfVars.LLAP_DAEMON_CONTAINER_ID.varname);
+        // Note, we assume production LLAP always runs under YARN.
+        LOG.error("Cannot find " + ApplicationConstants.Environment.CONTAINER_ID.toString()
+            + "; LLAP tokens may grant access to subsequent instances of the cluster with"
+            + " the same name");
+        appName = null;
       }
 
       int numExecutors = HiveConf.getIntVar(daemonConf, ConfVars.LLAP_DAEMON_NUM_EXECUTORS);
@@ -400,7 +426,8 @@ public class LlapDaemon extends CompositeService implements ContainerRunner, Lla
       boolean isDirectCache = HiveConf.getBoolVar(daemonConf, ConfVars.LLAP_ALLOCATOR_DIRECT);
       boolean isLlapIo = HiveConf.getBoolVar(daemonConf, HiveConf.ConfVars.LLAP_IO_ENABLED, true);
       llapDaemon = new LlapDaemon(daemonConf, numExecutors, executorMemoryBytes, isLlapIo,
-              isDirectCache, ioMemoryBytes, localDirs, rpcPort, mngPort, shufflePort, webPort);
+          isDirectCache, ioMemoryBytes, localDirs, rpcPort, mngPort, shufflePort, webPort,
+          appName);
 
       LOG.info("Adding shutdown hook for LlapDaemon");
       ShutdownHookManager.addShutdownHook(new CompositeServiceShutdownHook(llapDaemon), 1);
@@ -420,24 +447,27 @@ public class LlapDaemon extends CompositeService implements ContainerRunner, Lla
   }
 
   @Override
-  public SubmitWorkResponseProto submitWork(SubmitWorkRequestProto request) throws
-      IOException {
+  public SubmitWorkResponseProto submitWork(
+      SubmitWorkRequestProto request) throws IOException {
     numSubmissions.incrementAndGet();
     return containerRunner.submitWork(request);
   }
 
   @Override
-  public SourceStateUpdatedResponseProto sourceStateUpdated(SourceStateUpdatedRequestProto request) {
+  public SourceStateUpdatedResponseProto sourceStateUpdated(
+      SourceStateUpdatedRequestProto request) throws IOException {
     return containerRunner.sourceStateUpdated(request);
   }
 
   @Override
-  public QueryCompleteResponseProto queryComplete(QueryCompleteRequestProto request) {
+  public QueryCompleteResponseProto queryComplete(
+      QueryCompleteRequestProto request) throws IOException {
     return containerRunner.queryComplete(request);
   }
 
   @Override
-  public TerminateFragmentResponseProto terminateFragment(TerminateFragmentRequestProto request) {
+  public TerminateFragmentResponseProto terminateFragment(
+      TerminateFragmentRequestProto request) throws IOException {
     return containerRunner.terminateFragment(request);
   }
 

http://git-wip-us.apache.org/repos/asf/hive/blob/868e5e14/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
index dae1a3a..db8bfa6 100644
--- 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
@@ -32,6 +32,7 @@ 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.DaemonId;
 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;
@@ -71,13 +72,11 @@ public class LlapProtocolServerImpl extends AbstractService
   private final AtomicReference<InetSocketAddress> srvAddress, mngAddress;
   private SecretManager zkSecretManager;
   private String restrictedToUser = null;
+  private final DaemonId daemonId;
 
-  public LlapProtocolServerImpl(int numHandlers,
-                                ContainerRunner containerRunner,
-                                AtomicReference<InetSocketAddress> srvAddress,
-                                AtomicReference<InetSocketAddress> mngAddress,
-                                int srvPort,
-                                int mngPort) {
+  public LlapProtocolServerImpl(int numHandlers, ContainerRunner containerRunner,
+      AtomicReference<InetSocketAddress> srvAddress, AtomicReference<InetSocketAddress> mngAddress,
+      int srvPort, int mngPort, DaemonId daemonId) {
     super("LlapDaemonProtocolServerImpl");
     this.numHandlers = numHandlers;
     this.containerRunner = containerRunner;
@@ -85,14 +84,14 @@ public class LlapProtocolServerImpl extends AbstractService
     this.srvPort = srvPort;
     this.mngAddress = mngAddress;
     this.mngPort = mngPort;
+    this.daemonId = daemonId;
     LOG.info("Creating: " + LlapProtocolServerImpl.class.getSimpleName() +
         " with port configured to: " + srvPort);
   }
 
   @Override
   public SubmitWorkResponseProto submitWork(RpcController controller,
-                                            SubmitWorkRequestProto request) throws
-      ServiceException {
+      SubmitWorkRequestProto request) throws ServiceException {
     try {
       return containerRunner.submitWork(request);
     } catch (IOException e) {
@@ -103,20 +102,31 @@ public class LlapProtocolServerImpl extends AbstractService
   @Override
   public SourceStateUpdatedResponseProto sourceStateUpdated(RpcController controller,
       SourceStateUpdatedRequestProto request) throws ServiceException {
-    return containerRunner.sourceStateUpdated(request);
+    try {
+      return containerRunner.sourceStateUpdated(request);
+    } catch (IOException e) {
+      throw new ServiceException(e);
+    }
   }
 
   @Override
   public QueryCompleteResponseProto queryComplete(RpcController controller,
       QueryCompleteRequestProto request) throws ServiceException {
-    return containerRunner.queryComplete(request);
+    try {
+      return containerRunner.queryComplete(request);
+    } catch (IOException e) {
+      throw new ServiceException(e);
+    }
   }
 
   @Override
   public TerminateFragmentResponseProto terminateFragment(
-      RpcController controller,
-      TerminateFragmentRequestProto request) throws ServiceException {
-    return containerRunner.terminateFragment(request);
+      RpcController controller, TerminateFragmentRequestProto request) throws ServiceException {
+    try {
+      return containerRunner.terminateFragment(request);
+    } catch (IOException e) {
+      throw new ServiceException(e);
+    }
   }
 
   @Override
@@ -140,7 +150,7 @@ public class LlapProtocolServerImpl extends AbstractService
     }
     String llapPrincipal = HiveConf.getVar(conf, ConfVars.LLAP_KERBEROS_PRINCIPAL),
         llapKeytab = HiveConf.getVar(conf, ConfVars.LLAP_KERBEROS_KEYTAB_FILE);
-    zkSecretManager = SecretManager.createSecretManager(conf, llapPrincipal, llapKeytab);
+    zkSecretManager = SecretManager.createSecretManager(conf, llapPrincipal, llapKeytab, daemonId);
 
     // Start the protocol server after properly authenticating with daemon keytab.
     UserGroupInformation daemonUgi = null;
@@ -275,7 +285,8 @@ public class LlapProtocolServerImpl extends AbstractService
       realUser = new Text(ugi.getRealUser().getUserName());
     }
     Text renewer = new Text(ugi.getShortUserName());
-    LlapTokenIdentifier llapId = new LlapTokenIdentifier(owner, renewer, realUser);
+    LlapTokenIdentifier llapId = new LlapTokenIdentifier(owner, renewer, realUser,
+        daemonId.getClusterString(), request.hasAppId() ? request.getAppId() : null);
     // 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);
     if (LOG.isInfoEnabled()) {

http://git-wip-us.apache.org/repos/asf/hive/blob/868e5e14/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
new file mode 100644
index 0000000..03ee055
--- /dev/null
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/LlapTokenChecker.java
@@ -0,0 +1,137 @@
+/*
+ * 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 com.google.common.annotations.VisibleForTesting;
+
+import java.util.ArrayList;
+
+import java.util.List;
+
+import java.io.IOException;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.tuple.ImmutablePair;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.hadoop.hive.llap.security.LlapTokenIdentifier;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.security.token.TokenIdentifier;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public final class LlapTokenChecker {
+  private static final Logger LOG = LoggerFactory.getLogger(LlapTokenChecker.class);
+
+  private static final ImmutablePair<String, String> NO_SECURITY = new ImmutablePair<>(null, null);
+  public static Pair<String, String> getTokenInfo(String clusterId) throws IOException {
+    if (!UserGroupInformation.isSecurityEnabled()) return NO_SECURITY;
+    UserGroupInformation current = UserGroupInformation.getCurrentUser();
+    String kerberosName = current.hasKerberosCredentials() ? current.getShortUserName() : null;
+    List<LlapTokenIdentifier> tokens = getLlapTokens(current, clusterId);
+    if ((tokens == null || tokens.isEmpty()) && kerberosName == null) {
+      throw new SecurityException("No tokens or kerberos for " + current);
+    }
+    return getTokenInfoInternal(kerberosName, tokens);
+  }
+
+  private static List<LlapTokenIdentifier> getLlapTokens(
+      UserGroupInformation ugi, String clusterId) {
+    List<LlapTokenIdentifier> tokens = null;
+    for (TokenIdentifier id : ugi.getTokenIdentifiers()) {
+      if (!LlapTokenIdentifier.KIND_NAME.equals(id.getKind())) continue;
+      if (LOG.isDebugEnabled()) {
+        LOG.debug("Token {}", id);
+      }
+      LlapTokenIdentifier llapId = (LlapTokenIdentifier)id;
+      if (!clusterId.equals(llapId.getClusterId())) continue;
+      if (tokens == null) {
+        tokens = new ArrayList<>();
+      }
+      tokens.add((LlapTokenIdentifier)id);
+    }
+    return tokens;
+  }
+
+  @VisibleForTesting
+  static Pair<String, String> getTokenInfoInternal(
+      String kerberosName, List<LlapTokenIdentifier> tokens) {
+    assert (tokens != null && !tokens.isEmpty()) || kerberosName != null;
+    if (tokens == null) {
+      return new ImmutablePair<String, String>(kerberosName, null);
+    }
+    String userName = kerberosName, appId = null;
+    for (LlapTokenIdentifier llapId : tokens) {
+      String newUserName = llapId.getRealUser().toString();
+      if (userName != null && !userName.equals(newUserName)) {
+        throw new SecurityException("Ambiguous user name from credentials - " + userName
+            + " and " + newUserName + " from " + llapId
+            + ((kerberosName == null) ? ("; has kerberos credentials for " + kerberosName) : ""));
+      }
+      userName = newUserName;
+      String newAppId = llapId.getAppId();
+      if (!StringUtils.isEmpty(newAppId)) {
+        if (!StringUtils.isEmpty(appId) && !appId.equals(newAppId)) {
+          throw new SecurityException("Ambiguous app ID from credentials - " + appId
+              + " and " + newAppId + " from " + llapId);
+        }
+        appId = newAppId;
+      }
+    }
+    assert userName != null;
+    return new ImmutablePair<String, String>(userName, appId);
+  }
+
+  public static void checkPermissions(
+      String clusterId, String userName, String appId, Object hint) throws IOException {
+    if (!UserGroupInformation.isSecurityEnabled()) return;
+    UserGroupInformation current = UserGroupInformation.getCurrentUser();
+    String kerberosName = current.hasKerberosCredentials() ? current.getShortUserName() : null;
+    List<LlapTokenIdentifier> tokens = getLlapTokens(current, clusterId);
+    checkPermissionsInternal(kerberosName, tokens, userName, appId, hint);
+  }
+
+  @VisibleForTesting
+  static void checkPermissionsInternal(String kerberosName, List<LlapTokenIdentifier> tokens,
+      String userName, String appId, Object hint) {
+    if (kerberosName != null && StringUtils.isEmpty(appId) && kerberosName.equals(userName)) {
+      return;
+    }
+    if (tokens != null) {
+      for (LlapTokenIdentifier llapId : tokens) {
+        String tokenUser = llapId.getRealUser().toString(), tokenAppId = llapId.getAppId();
+        if (checkTokenPermissions(userName, appId, tokenUser, tokenAppId)) return;
+      }
+    }
+    throw new SecurityException("Unauthorized to access "
+        + userName + ", " + appId.hashCode() + " (" + hint + ")");
+  }
+
+  public static void checkPermissions(
+      Pair<String, String> prm, String userName, String appId, Object hint) {
+    if (userName == null) {
+      assert StringUtils.isEmpty(appId);
+      return;
+    }
+    if (!checkTokenPermissions(userName, appId, prm.getLeft(), prm.getRight())) {
+      throw new SecurityException("Unauthorized to access "
+          + userName + ", " + appId.hashCode() + " (" + hint + ")");
+    }
+  }
+
+  private static boolean checkTokenPermissions(
+      String userName, String appId, String tokenUser, String tokenAppId) {
+    return userName.equals(tokenUser)
+        && (StringUtils.isEmpty(appId) || appId.equals(tokenAppId));
+  }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/hive/blob/868e5e14/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/QueryInfo.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/QueryInfo.java b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/QueryInfo.java
index 64c2b58..8daef9e 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/QueryInfo.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/QueryInfo.java
@@ -55,10 +55,11 @@ public class QueryInfo {
   private final ConcurrentMap<String, SourceStateProto> sourceStateMap;
 
   private final FinishableStateTracker finishableStateTracker = new FinishableStateTracker();
+  private final String tokenUserName, appId;
 
-  public QueryInfo(QueryIdentifier queryIdentifier, String appIdString, String dagName, int dagIdentifier,
-                   String user, ConcurrentMap<String, SourceStateProto> sourceStateMap,
-                   String[] localDirsBase, FileSystem localFs) {
+  public QueryInfo(QueryIdentifier queryIdentifier, String appIdString, String dagName,
+      int dagIdentifier, String user, ConcurrentMap<String, SourceStateProto> sourceStateMap,
+      String[] localDirsBase, FileSystem localFs, String tokenUserName, String tokenAppId) {
     this.queryIdentifier = queryIdentifier;
     this.appIdString = appIdString;
     this.dagName = dagName;
@@ -67,6 +68,8 @@ public class QueryInfo {
     this.user = user;
     this.localDirsBase = localDirsBase;
     this.localFs = localFs;
+    this.tokenUserName = tokenUserName;
+    this.appId = tokenAppId;
   }
 
   public QueryIdentifier getQueryIdentifier() {
@@ -270,4 +273,12 @@ public class QueryInfo {
       this.lastFinishableState = lastFinishableState;
     }
   }
+
+  public String getTokenUserName() {
+    return tokenUserName;
+  }
+
+  public String getTokenAppId() {
+    return appId;
+  }
 }

http://git-wip-us.apache.org/repos/asf/hive/blob/868e5e14/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/QueryTracker.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/QueryTracker.java b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/QueryTracker.java
index 14657e6..cb3be2b 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/QueryTracker.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/QueryTracker.java
@@ -18,6 +18,8 @@ import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
+
+import org.apache.commons.lang3.tuple.Pair;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.security.token.Token;
 import org.apache.tez.common.CallableWithNdc;
@@ -60,6 +62,7 @@ public class QueryTracker extends AbstractService {
 
   private final String[] localDirsBase;
   private final FileSystem localFs;
+  private final String clusterId;
   private final long defaultDeleteDelaySeconds;
 
   // TODO At the moment there's no way of knowing whether a query is running or not.
@@ -89,9 +92,10 @@ public class QueryTracker extends AbstractService {
   private final ConcurrentHashMap<QueryIdentifier, String> queryIdentifierToHiveQueryId =
       new ConcurrentHashMap<>();
 
-  public QueryTracker(Configuration conf, String[] localDirsBase) {
+  public QueryTracker(Configuration conf, String[] localDirsBase, String clusterId) {
     super("QueryTracker");
     this.localDirsBase = localDirsBase;
+    this.clusterId = clusterId;
     try {
       localFs = FileSystem.getLocal(conf);
     } catch (IOException e) {
@@ -119,35 +123,50 @@ public class QueryTracker extends AbstractService {
    * @param user
    * @throws IOException
    */
-  QueryFragmentInfo registerFragment(QueryIdentifier queryIdentifier, String appIdString, String dagName,
-                                     int dagIdentifier, String vertexName, int fragmentNumber, int attemptNumber, String user,
-                                     FragmentSpecProto fragmentSpec, Token<JobTokenIdentifier> appToken) throws IOException {
+  QueryFragmentInfo registerFragment(QueryIdentifier queryIdentifier, String appIdString,
+      String dagName, int dagIdentifier, String vertexName, int fragmentNumber, int attemptNumber,
+      String user, FragmentSpecProto fragmentSpec, Token<JobTokenIdentifier> appToken)
+          throws IOException {
     ReadWriteLock dagLock = getDagLock(queryIdentifier);
     dagLock.readLock().lock();
     try {
-      if (!completedDagMap.contains(queryIdentifier)) {
-        QueryInfo queryInfo = queryInfoMap.get(queryIdentifier);
-        if (queryInfo == null) {
-          queryInfo = new QueryInfo(queryIdentifier, appIdString, dagName, dagIdentifier, user,
-              getSourceCompletionMap(queryIdentifier), localDirsBase, localFs);
-          queryInfoMap.putIfAbsent(queryIdentifier, queryInfo);
-        }
-
-        if (LOG.isDebugEnabled()) {
-          LOG.debug("Registering request for {} with the ShuffleHandler", queryIdentifier);
-        }
-        ShuffleHandler.get()
-            .registerDag(appIdString, dagIdentifier, appToken,
-                user, queryInfo.getLocalDirs());
-
-        return queryInfo.registerFragment(vertexName, fragmentNumber, attemptNumber, fragmentSpec);
-      } else {
+      if (completedDagMap.contains(queryIdentifier)) {
         // Cleanup the dag lock here, since it may have been created after the query completed
         dagSpecificLocks.remove(queryIdentifier);
         throw new RuntimeException(
             "Dag " + dagName + " already complete. Rejecting fragment ["
                 + vertexName + ", " + fragmentNumber + ", " + attemptNumber + "]");
       }
+      // TODO: for now, we get the secure username out of UGI... after signing, we can take it
+      //       out of the request provided that it's signed.
+      Pair<String, String> tokenInfo = LlapTokenChecker.getTokenInfo(clusterId);
+      boolean isExistingQueryInfo = true;
+      QueryInfo queryInfo = queryInfoMap.get(queryIdentifier);
+      if (queryInfo == null) {
+        queryInfo = new QueryInfo(queryIdentifier, appIdString, dagName, dagIdentifier, user,
+            getSourceCompletionMap(queryIdentifier), localDirsBase, localFs,
+            tokenInfo.getLeft(), tokenInfo.getRight());
+        QueryInfo old = queryInfoMap.putIfAbsent(queryIdentifier, queryInfo);
+        if (old != null) {
+          queryInfo = old;
+        } else {
+          isExistingQueryInfo = false;
+        }
+      }
+      if (isExistingQueryInfo) {
+        // We already retrieved the incoming info, check without UGI.
+        LlapTokenChecker.checkPermissions(tokenInfo, queryInfo.getTokenUserName(),
+            queryInfo.getTokenAppId(), queryInfo.getQueryIdentifier());
+      }
+
+      if (LOG.isDebugEnabled()) {
+        LOG.debug("Registering request for {} with the ShuffleHandler", queryIdentifier);
+      }
+      ShuffleHandler.get()
+          .registerDag(appIdString, dagIdentifier, appToken,
+              user, queryInfo.getLocalDirs());
+
+      return queryInfo.registerFragment(vertexName, fragmentNumber, attemptNumber, fragmentSpec);
     } finally {
       dagLock.readLock().unlock();
     }
@@ -174,17 +193,20 @@ public class QueryTracker extends AbstractService {
    * @param queryIdentifier
    * @param deleteDelay
    */
-  List<QueryFragmentInfo> queryComplete(QueryIdentifier queryIdentifier, long deleteDelay) {
+  List<QueryFragmentInfo> queryComplete(QueryIdentifier queryIdentifier, long deleteDelay,
+      boolean isInternal) throws IOException {
     if (deleteDelay == -1) {
       deleteDelay = defaultDeleteDelaySeconds;
     }
     ReadWriteLock dagLock = getDagLock(queryIdentifier);
     dagLock.writeLock().lock();
     try {
+      QueryInfo queryInfo = isInternal
+          ? queryInfoMap.get(queryIdentifier) : checkPermissionsAndGetQuery(queryIdentifier);
       rememberCompletedDag(queryIdentifier);
       LOG.info("Processing queryComplete for queryIdentifier={} with deleteDelay={} seconds", queryIdentifier,
           deleteDelay);
-      QueryInfo queryInfo = queryInfoMap.remove(queryIdentifier);
+      queryInfoMap.remove(queryIdentifier);
       if (queryInfo == null) {
         LOG.warn("Ignoring query complete for unknown dag: {}", queryIdentifier);
         return Collections.emptyList();
@@ -229,9 +251,10 @@ public class QueryTracker extends AbstractService {
    * @param sourceName
    * @param sourceState
    */
-  void registerSourceStateChange(QueryIdentifier queryIdentifier, String sourceName, SourceStateProto sourceState) {
+  void registerSourceStateChange(QueryIdentifier queryIdentifier, String sourceName,
+      SourceStateProto sourceState) throws IOException {
     getSourceCompletionMap(queryIdentifier).put(sourceName, sourceState);
-    QueryInfo queryInfo = queryInfoMap.get(queryIdentifier);
+    QueryInfo queryInfo = checkPermissionsAndGetQuery(queryIdentifier);
     if (queryInfo != null) {
       queryInfo.sourceStateUpdated(sourceName);
     } else {
@@ -322,4 +345,16 @@ public class QueryTracker extends AbstractService {
       return null;
     }
   }
+
+  private QueryInfo checkPermissionsAndGetQuery(QueryIdentifier queryId) throws IOException {
+    QueryInfo queryInfo = queryInfoMap.get(queryId);
+    if (queryInfo == null) return null;
+    LlapTokenChecker.checkPermissions(clusterId, queryInfo.getTokenAppId(),
+        queryInfo.getTokenUserName(), queryInfo.getQueryIdentifier());
+    return queryInfo;
+  }
+
+  public boolean checkPermissionsForQuery(QueryIdentifier queryId) throws IOException {
+    return checkPermissionsAndGetQuery(queryId) != null;
+  }
 }

http://git-wip-us.apache.org/repos/asf/hive/blob/868e5e14/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/Scheduler.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/Scheduler.java b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/Scheduler.java
index 26c8e55..fd6234a 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/Scheduler.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/Scheduler.java
@@ -44,4 +44,6 @@ public interface Scheduler<T> {
   void killFragment(String fragmentId);
 
   Set<String> getExecutorsStatus();
+
+  QueryIdentifier findQueryByFragment(String fragmentId);
 }

http://git-wip-us.apache.org/repos/asf/hive/blob/868e5e14/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/TaskExecutorService.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/TaskExecutorService.java b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/TaskExecutorService.java
index f621af2..1933eb1 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/TaskExecutorService.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/TaskExecutorService.java
@@ -420,6 +420,15 @@ public class TaskExecutorService extends AbstractService implements Scheduler<Ta
   }
 
   @Override
+  public QueryIdentifier findQueryByFragment(String fragmentId) {
+    synchronized (lock) {
+      TaskWrapper taskWrapper = knownTasks.get(fragmentId);
+      return taskWrapper == null ? null : taskWrapper.getTaskRunnerCallable()
+          .getFragmentInfo().getQueryInfo().getQueryIdentifier();
+    }
+  }
+
+  @Override
   public void killFragment(String fragmentId) {
     synchronized (lock) {
       TaskWrapper taskWrapper = knownTasks.remove(fragmentId);

http://git-wip-us.apache.org/repos/asf/hive/blob/868e5e14/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 76ba225..f958bc4 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
@@ -26,6 +26,7 @@ import java.util.concurrent.TimeUnit;
 
 import javax.net.SocketFactory;
 
+import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hive.llap.impl.LlapManagementProtocolClientImpl;
 import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.GetTokenRequestProto;
@@ -81,7 +82,7 @@ public class LlapSecurityHelper implements LlapTokenProvider {
   }
 
   @Override
-  public Token<LlapTokenIdentifier> getDelegationToken() throws IOException {
+  public Token<LlapTokenIdentifier> getDelegationToken(String appId) throws IOException {
     if (!UserGroupInformation.isSecurityEnabled()) return null;
     if (llapUgi == null) {
       llapUgi = UserGroupInformation.getCurrentUser();
@@ -98,7 +99,7 @@ public class LlapSecurityHelper implements LlapTokenProvider {
     boolean hasRefreshed = false;
     while (true) {
       try {
-        tokenBytes = getTokenBytes();
+        tokenBytes = getTokenBytes(appId);
         break;
       } catch (InterruptedException ie) {
         throw new RuntimeException(ie);
@@ -128,7 +129,8 @@ public class LlapSecurityHelper implements LlapTokenProvider {
     return token;
   }
 
-  private ByteString getTokenBytes() throws InterruptedException, IOException {
+  private ByteString getTokenBytes(
+      final String appId) throws InterruptedException, IOException {
     return llapUgi.doAs(new PrivilegedExceptionAction<ByteString>() {
       @Override
       public ByteString run() throws Exception {
@@ -138,8 +140,11 @@ public class LlapSecurityHelper implements LlapTokenProvider {
               clientInstance.getManagementPort(), retryPolicy, socketFactory);
         }
         // Client only connects on the first call, so this has to be done in doAs.
-        GetTokenRequestProto req = GetTokenRequestProto.newBuilder().build();
-        return client.getDelegationToken(null, req).getToken();
+        GetTokenRequestProto.Builder req = GetTokenRequestProto.newBuilder();
+        if (!StringUtils.isBlank(appId)) {
+          req.setAppId(appId);
+        }
+        return client.getDelegationToken(null, req.build()).getToken();
       }
     });
   }

http://git-wip-us.apache.org/repos/asf/hive/blob/868e5e14/llap-server/src/java/org/apache/hadoop/hive/llap/security/SecretManager.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/security/SecretManager.java b/llap-server/src/java/org/apache/hadoop/hive/llap/security/SecretManager.java
index 8c7a539..c54e726 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/security/SecretManager.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/security/SecretManager.java
@@ -23,6 +23,7 @@ import java.util.regex.Pattern;
 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.DaemonId;
 import org.apache.hadoop.hive.llap.security.LlapTokenIdentifier;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.security.token.Token;
@@ -33,6 +34,7 @@ import org.slf4j.LoggerFactory;
 
 public class SecretManager extends ZKDelegationTokenSecretManager<LlapTokenIdentifier> {
   private static final Logger LOG = LoggerFactory.getLogger(SecretManager.class);
+
   public SecretManager(Configuration conf) {
     super(conf);
     checkForZKDTSMBug(conf);
@@ -82,16 +84,8 @@ public class SecretManager extends ZKDelegationTokenSecretManager<LlapTokenIdent
     return id;
   }
 
-  private final static Pattern hostsRe = Pattern.compile("[^A-Za-z0-9_-]");
-  private static String deriveZkPath(Configuration conf) throws IOException {
-    String hosts = HiveConf.getTrimmedVar(conf, ConfVars.LLAP_DAEMON_SERVICE_HOSTS);
-    String clusterName = hosts.startsWith("@") ? hosts.substring(1) : hosts;
-    String userName = UserGroupInformation.getCurrentUser().getShortUserName();
-    return hostsRe.matcher(userName + "_" + clusterName).replaceAll("_") ;
-  }
-
   public static SecretManager createSecretManager(
-      final Configuration conf, String llapPrincipal, String llapKeytab) {
+      final Configuration conf, String llapPrincipal, String llapKeytab, DaemonId daemonId) {
     // Create ZK connection under a separate ugi (if specified) - ZK works in mysterious ways.
     UserGroupInformation zkUgi = null;
     String principal = HiveConf.getVar(conf, ConfVars.LLAP_ZKSM_KERBEROS_PRINCIPAL, llapPrincipal);
@@ -110,12 +104,7 @@ public class SecretManager extends ZKDelegationTokenSecretManager<LlapTokenIdent
     zkConf.setLong(DelegationTokenManager.RENEW_INTERVAL, tokenLifetime);
     zkConf.set(SecretManager.ZK_DTSM_ZK_KERBEROS_PRINCIPAL, principal);
     zkConf.set(SecretManager.ZK_DTSM_ZK_KERBEROS_KEYTAB, keyTab);
-    String zkPath;
-    try {
-      zkPath = deriveZkPath(conf);
-    } catch (IOException e) {
-      throw new RuntimeException(e);
-    }
+    String zkPath = daemonId.getClusterString();
     LOG.info("Using {} as ZK secret manager path", zkPath);
     zkConf.set(SecretManager.ZK_DTSM_ZNODE_WORKING_PATH, "zkdtsm_" + zkPath);
     setZkConfIfNotSet(zkConf, SecretManager.ZK_DTSM_ZK_AUTH_TYPE, "sasl");

http://git-wip-us.apache.org/repos/asf/hive/blob/868e5e14/llap-server/src/test/org/apache/hadoop/hive/llap/daemon/MiniLlapCluster.java
----------------------------------------------------------------------
diff --git a/llap-server/src/test/org/apache/hadoop/hive/llap/daemon/MiniLlapCluster.java b/llap-server/src/test/org/apache/hadoop/hive/llap/daemon/MiniLlapCluster.java
index 610f266..dde5be0 100644
--- a/llap-server/src/test/org/apache/hadoop/hive/llap/daemon/MiniLlapCluster.java
+++ b/llap-server/src/test/org/apache/hadoop/hive/llap/daemon/MiniLlapCluster.java
@@ -193,7 +193,7 @@ public class MiniLlapCluster extends AbstractService {
     LOG.info("Initializing {} llap instances for MiniLlapCluster with name={}", numInstances, clusterNameTrimmed);
     for (int i = 0 ;i < numInstances ; i++) {
       llapDaemons[i] = new LlapDaemon(conf, numExecutorsPerService, execBytesPerService, llapIoEnabled,
-          ioIsDirect, ioBytesPerService, localDirs, rpcPort, mngPort, shufflePort, webPort);
+          ioIsDirect, ioBytesPerService, localDirs, rpcPort, mngPort, shufflePort, webPort, clusterNameTrimmed);
       llapDaemons[i].init(new Configuration(conf));
     }
     LOG.info("Initialized {} llap instances for MiniLlapCluster with name={}", numInstances, clusterNameTrimmed);

http://git-wip-us.apache.org/repos/asf/hive/blob/868e5e14/llap-server/src/test/org/apache/hadoop/hive/llap/daemon/impl/TaskExecutorTestHelpers.java
----------------------------------------------------------------------
diff --git a/llap-server/src/test/org/apache/hadoop/hive/llap/daemon/impl/TaskExecutorTestHelpers.java b/llap-server/src/test/org/apache/hadoop/hive/llap/daemon/impl/TaskExecutorTestHelpers.java
index 24f4442..c6ba14e 100644
--- a/llap-server/src/test/org/apache/hadoop/hive/llap/daemon/impl/TaskExecutorTestHelpers.java
+++ b/llap-server/src/test/org/apache/hadoop/hive/llap/daemon/impl/TaskExecutorTestHelpers.java
@@ -78,7 +78,7 @@ public class TaskExecutorTestHelpers {
     QueryInfo queryInfo =
         new QueryInfo(queryIdentifier, "fake_app_id_string", "fake_dag_name", 1, "fakeUser",
             new ConcurrentHashMap<String, LlapDaemonProtocolProtos.SourceStateProto>(),
-            new String[0], null);
+            new String[0], null, null, null);
     return queryInfo;
   }
 

http://git-wip-us.apache.org/repos/asf/hive/blob/868e5e14/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 a65bf5c..fd37a06 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
@@ -46,7 +46,7 @@ public class TestLlapDaemonProtocolServerImpl {
     LlapProtocolServerImpl server =
         new LlapProtocolServerImpl(numHandlers, containerRunnerMock,
            new AtomicReference<InetSocketAddress>(), new AtomicReference<InetSocketAddress>(),
-           rpcPort, rpcPort + 1);
+           rpcPort, rpcPort + 1, null);
     when(containerRunnerMock.submitWork(any(SubmitWorkRequestProto.class))).thenReturn(
         SubmitWorkResponseProto
             .newBuilder()

http://git-wip-us.apache.org/repos/asf/hive/blob/868e5e14/llap-server/src/test/org/apache/hadoop/hive/llap/daemon/impl/TestLlapTokenChecker.java
----------------------------------------------------------------------
diff --git a/llap-server/src/test/org/apache/hadoop/hive/llap/daemon/impl/TestLlapTokenChecker.java b/llap-server/src/test/org/apache/hadoop/hive/llap/daemon/impl/TestLlapTokenChecker.java
new file mode 100644
index 0000000..aaaa762
--- /dev/null
+++ b/llap-server/src/test/org/apache/hadoop/hive/llap/daemon/impl/TestLlapTokenChecker.java
@@ -0,0 +1,96 @@
+/**
+ * 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.daemon.impl;
+
+import static org.junit.Assert.*;
+
+import org.apache.hadoop.io.Text;
+
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.hadoop.hive.llap.security.LlapTokenIdentifier;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.junit.Test;
+
+public class TestLlapTokenChecker {
+
+  @Test
+  public void testGetToken() {
+    check(LlapTokenChecker.getTokenInfoInternal("u", null), "u", null);
+    check(LlapTokenChecker.getTokenInfoInternal(null, createTokens("u", null)), "u", null);
+    check(LlapTokenChecker.getTokenInfoInternal(null, createTokens("u", "a")), "u", "a");
+    check(LlapTokenChecker.getTokenInfoInternal("u", createTokens("u", "a")), "u", "a");
+    check(LlapTokenChecker.getTokenInfoInternal("u", createTokens("u", "a", "u", null)),
+        "u", "a");
+    // Note - some of these scenarios could be handled, but they are not supported right now.
+    // The reason is that we bind a query to app/user using the signed token information, and
+    // we don't want to bother figuring out which one to use in case of ambiguity w/o a use case.
+    checkGetThrows("u", createTokens("u2", "a")); // Ambiguous user.
+    checkGetThrows("u2", createTokens("u2", "a", "u3", "a")); // Ambiguous user.
+    checkGetThrows(null, createTokens("u2", "a", "u3", "a")); // Ambiguous user.
+    checkGetThrows(null, createTokens("u2", "a", "u2", "a1")); // Ambiguous app.
+  }
+
+  @Test
+  public void testCheckPermissions() {
+    LlapTokenChecker.checkPermissionsInternal("u", null, "u", null, null);
+    LlapTokenChecker.checkPermissionsInternal(null, createTokens("u", null) , "u", null, null);
+    LlapTokenChecker.checkPermissionsInternal("u", createTokens("u", "a") , "u", "a", null);
+    // No access.
+    checkPrmThrows("u2", null, "u", "a");
+    checkPrmThrows("u", null, "u", "a"); // Note - Kerberos user w/o appId doesn't have access.
+    checkPrmThrows(null, createTokens("u2", "a"), "u", "a");
+    checkPrmThrows(null, createTokens("u", "a2"), "u", "a");
+    checkPrmThrows(null, createTokens("u", null), "u", "a");
+  }
+
+  private List<LlapTokenIdentifier> createTokens(String... args) {
+    List<LlapTokenIdentifier> tokens = new ArrayList<>();
+    for (int i = 0; i < args.length; i += 2) {
+      tokens.add(new LlapTokenIdentifier(null, null, new Text(args[i]), "c", args[i + 1]));
+    }
+    return tokens;
+  }
+
+  private void checkGetThrows(String kerberosName, List<LlapTokenIdentifier> tokens) {
+    try {
+      LlapTokenChecker.getTokenInfoInternal(kerberosName, tokens);
+      fail("Didn't throw");
+    } catch (SecurityException ex) {
+      // Expected.
+    }
+  }
+
+  private void checkPrmThrows(
+      String kerberosName, List<LlapTokenIdentifier> tokens, String userName, String appId) {
+    try {
+      LlapTokenChecker.checkPermissionsInternal(kerberosName, tokens, userName, appId, null);
+      fail("Didn't throw");
+    } catch (SecurityException ex) {
+      // Expected.
+    }
+  }
+
+  private void check(Pair<String, String> p, String user, String appId) {
+    assertEquals(user, p.getLeft());
+    assertEquals(appId, p.getRight());
+  }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/hive/blob/868e5e14/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezSessionState.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezSessionState.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezSessionState.java
index 3ea5ef9..fd6465a 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezSessionState.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezSessionState.java
@@ -275,7 +275,8 @@ public class TezSessionState {
     if (llapMode) {
       if (UserGroupInformation.isSecurityEnabled()) {
         LlapTokenProvider tp = LlapProxy.getOrInitTokenProvider(conf);
-        Token<LlapTokenIdentifier> token = tp.getDelegationToken();
+        // For Tez, we don't use appId to distinguish the tokens; security scope is the user.
+        Token<LlapTokenIdentifier> token = tp.getDelegationToken(null);
         if (LOG.isInfoEnabled()) {
           LOG.info("Obtained a LLAP token: " + token);
         }


Mime
View raw message