Repository: hive
Updated Branches:
  refs/heads/master 43837e8ef -> 1e6fa1eb3


HIVE-12931. Shuffle tokens stay around forever in LLAP. (Siddharth Seth, 
reviewed by Sergey Shelukhin)


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

Branch: refs/heads/master
Commit: 1e6fa1eb3f8b7eeb551355cb6ebee483ef4e4d71
Parents: 43837e8
Author: Siddharth Seth <ss...@apache.org>
Authored: Mon Feb 1 11:14:37 2016 -0800
Committer: Siddharth Seth <ss...@apache.org>
Committed: Mon Feb 1 11:14:37 2016 -0800

----------------------------------------------------------------------
 .../llap/daemon/impl/ContainerRunnerImpl.java   | 31 +++++--------
 .../hive/llap/daemon/impl/QueryTracker.java     | 14 +++++-
 .../llap/shufflehandler/ShuffleHandler.java     | 49 +++++++++++++++-----
 3 files changed, 62 insertions(+), 32 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hive/blob/1e6fa1eb/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 535fe76..e80fb15 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
@@ -46,7 +46,6 @@ import 
org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.SubmitWor
 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.metrics.LlapDaemonExecutorMetrics;
-import org.apache.hadoop.hive.llap.shufflehandler.ShuffleHandler;
 import org.apache.hadoop.hive.ql.exec.tez.TezProcessor;
 import org.apache.hadoop.io.DataInputBuffer;
 import org.apache.hadoop.security.Credentials;
@@ -178,36 +177,30 @@ public class ContainerRunnerImpl extends CompositeService 
implements ContainerRu
 
       QueryIdentifier queryIdentifier = new 
QueryIdentifier(request.getApplicationIdString(), dagIdentifier);
 
+      Credentials credentials = new Credentials();
+      DataInputBuffer dib = new DataInputBuffer();
+      byte[] tokenBytes = request.getCredentialsBinary().toByteArray();
+      dib.reset(tokenBytes, tokenBytes.length);
+      credentials.readTokenStorageStream(dib);
+
+      Token<JobTokenIdentifier> jobToken = 
TokenCache.getSessionToken(credentials);
+
       QueryFragmentInfo fragmentInfo = queryTracker
-          .registerFragment(queryIdentifier, request.getApplicationIdString(), 
fragmentSpec.getDagName(),
+          .registerFragment(queryIdentifier, request.getApplicationIdString(),
+              fragmentSpec.getDagName(),
               dagIdentifier,
               fragmentSpec.getVertexName(), fragmentSpec.getFragmentNumber(),
-              fragmentSpec.getAttemptNumber(), request.getUser(), 
request.getFragmentSpec());
+              fragmentSpec.getAttemptNumber(), request.getUser(), 
request.getFragmentSpec(),
+              jobToken);
 
       String[] localDirs = fragmentInfo.getLocalDirs();
       Preconditions.checkNotNull(localDirs);
-
       if (LOG.isDebugEnabled()) {
         LOG.debug("Dirs are: " + Arrays.toString(localDirs));
       }
       // 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)
 
-      Credentials credentials = new Credentials();
-      DataInputBuffer dib = new DataInputBuffer();
-      byte[] tokenBytes = request.getCredentialsBinary().toByteArray();
-      dib.reset(tokenBytes, tokenBytes.length);
-      credentials.readTokenStorageStream(dib);
-
-      Token<JobTokenIdentifier> jobToken = 
TokenCache.getSessionToken(credentials);
-
-      if (LOG.isDebugEnabled()) {
-        LOG.debug("Registering request with the ShuffleHandler");
-      }
-      ShuffleHandler.get()
-          .registerDag(request.getApplicationIdString(), dagIdentifier, 
jobToken,
-              request.getUser(), localDirs);
-
       TaskRunnerCallable callable = new TaskRunnerCallable(request, 
fragmentInfo, new Configuration(getConfig()),
           new LlapExecutionContext(localAddress.get().getHostName(), 
queryTracker), env,
           credentials, memoryPerExecutor, amReporter, confParams, metrics, 
killedTaskHandler,

http://git-wip-us.apache.org/repos/asf/hive/blob/1e6fa1eb/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 0676edd..80264a0 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
@@ -19,6 +19,7 @@ import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.security.token.Token;
 import org.apache.tez.common.CallableWithNdc;
 
 import org.apache.hadoop.service.AbstractService;
@@ -31,6 +32,7 @@ import 
org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.FragmentS
 import 
org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.SourceStateProto;
 import org.apache.hadoop.hive.llap.shufflehandler.ShuffleHandler;
 import org.apache.hadoop.hive.ql.exec.ObjectCacheFactory;
+import org.apache.tez.common.security.JobTokenIdentifier;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -120,8 +122,8 @@ public class QueryTracker extends AbstractService {
    * @throws IOException
    */
   QueryFragmentInfo registerFragment(QueryIdentifier queryIdentifier, String 
appIdString, String dagName,
-      int dagIdentifier, String vertexName, int fragmentNumber, int 
attemptNumber, String user,
-      FragmentSpecProto fragmentSpec) throws IOException {
+                                     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 {
@@ -132,6 +134,14 @@ public class QueryTracker extends AbstractService {
               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 {
         // Cleanup the dag lock here, since it may have been created after the 
query completed

http://git-wip-us.apache.org/repos/asf/hive/blob/1e6fa1eb/llap-server/src/java/org/apache/hadoop/hive/llap/shufflehandler/ShuffleHandler.java
----------------------------------------------------------------------
diff --git 
a/llap-server/src/java/org/apache/hadoop/hive/llap/shufflehandler/ShuffleHandler.java
 
b/llap-server/src/java/org/apache/hadoop/hive/llap/shufflehandler/ShuffleHandler.java
index b042455..2c51169 100644
--- 
a/llap-server/src/java/org/apache/hadoop/hive/llap/shufflehandler/ShuffleHandler.java
+++ 
b/llap-server/src/java/org/apache/hadoop/hive/llap/shufflehandler/ShuffleHandler.java
@@ -155,7 +155,7 @@ public class ShuffleHandler implements 
AttemptRegistrationListener {
   private final ReadaheadPool readaheadPool = ReadaheadPool.getInstance();
 
   /* List of registered applications */
-  private final ConcurrentMap<String, Boolean> registeredApps = new 
ConcurrentHashMap<String, Boolean>();
+  private final ConcurrentMap<String, Integer> registeredApps = new 
ConcurrentHashMap<>();
   /* Maps application identifiers (jobIds) to the associated user for the app 
*/
   private final ConcurrentMap<String,String> userRsrc;
   private JobTokenSecretManager secretManager;
@@ -406,6 +406,9 @@ public class ShuffleHandler implements 
AttemptRegistrationListener {
 
   /**
    * Register an application and it's associated credentials and user 
information.
+   *
+   * This method and unregisterDag must be synchronized externally to prevent 
races in shuffle token registration/unregistration
+   *
    * @param applicationIdString
    * @param dagIdentifier
    * @param appToken
@@ -414,12 +417,24 @@ public class ShuffleHandler implements 
AttemptRegistrationListener {
   public void registerDag(String applicationIdString, int dagIdentifier,
                           Token<JobTokenIdentifier> appToken,
                           String user, String[] appDirs) {
-    // TODO Fix this. There's a race here, where an app may think everything 
is registered, finish really fast, send events and the consumer will not find 
the registration.
-    Boolean registered = registeredApps.putIfAbsent(applicationIdString, 
Boolean.valueOf(true));
-    if (registered == null) {
-      LOG.debug("Registering watches for AppDirs: appId=" + 
applicationIdString);
+    Integer registeredDagIdentifier = 
registeredApps.putIfAbsent(applicationIdString, dagIdentifier);
+    // App never seen, or previous dag has been unregistered.
+    if (registeredDagIdentifier == null) {
       recordJobShuffleInfo(applicationIdString, user, appToken);
+    }
+    // Register the new dag identifier, if that's not the one currently 
registered.
+    // Register comes in before the unregister for the previous dag
+    if (registeredDagIdentifier != null && 
!registeredDagIdentifier.equals(dagIdentifier)) {
+      registeredApps.put(applicationIdString, dagIdentifier);
+      // Don't need to recordShuffleInfo since the out of sync unregister will 
not remove the
+      // credentials
+    }
+    // First time registration, or new register comes in before the previous 
unregister.
+    if (registeredDagIdentifier == null || 
!registeredDagIdentifier.equals(dagIdentifier)) {
       if (dirWatcher != null) {
+        if (LOG.isDebugEnabled()) {
+          LOG.debug("Registering watches for AppDirs: appId={}, dagId={}", 
applicationIdString, dagIdentifier);
+        }
         for (String appDir : appDirs) {
           try {
             dirWatcher.registerDagDir(appDir, applicationIdString, 
dagIdentifier, user,
@@ -432,15 +447,27 @@ public class ShuffleHandler implements 
AttemptRegistrationListener {
     }
   }
 
+  /**
+   * Unregister a specific dag
+   *
+   * This method and registerDag must be synchronized externally to prevent 
races in shuffle token registration/unregistration
+   *
+   * @param dir
+   * @param applicationIdString
+   * @param dagIdentifier
+   */
   public void unregisterDag(String dir, String applicationIdString, int 
dagIdentifier) {
+    Integer currentDagIdentifier = registeredApps.get(applicationIdString);
+    // Unregister may come in after the new dag has started running. The 
methods are expected to
+    // be synchronized, hence the following check is sufficient.
+    if (currentDagIdentifier != null && 
currentDagIdentifier.equals(dagIdentifier)) {
+      registeredApps.remove(applicationIdString);
+      removeJobShuffleInfo(applicationIdString);
+    }
+    // Unregister for the dirWatcher for the specific dagIdentifier in either 
case.
     if (dirWatcher != null) {
       dirWatcher.unregisterDagDir(dir, applicationIdString, dagIdentifier);
     }
-    // TODO Cleanup registered tokens and dag info
-  }
-
-  public void unregisterApplication(String applicationIdString) {
-    removeJobShuffleInfo(applicationIdString);
   }
 
 
@@ -468,7 +495,7 @@ public class ShuffleHandler implements 
AttemptRegistrationListener {
     // This is in place to be compatible with the MR ShuffleHandler. Requests 
from ShuffleInputs
     // arrive with a job_ prefix.
     String jobIdString = appIdString.replace("application", "job");
-    userRsrc.put(jobIdString, user);
+    userRsrc.putIfAbsent(jobIdString, user);
     secretManager.addTokenForJob(jobIdString, jobToken);
     LOG.info("Added token for " + jobIdString);
   }

Reply via email to