HIVE-10736: HiveServer2 shutdown of cached tez app-masters is not clean (Vikram Dixit K, reviewed by Gunther Hagleitner)
Project: http://git-wip-us.apache.org/repos/asf/hive/repo Commit: http://git-wip-us.apache.org/repos/asf/hive/commit/dbd89ac4 Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/dbd89ac4 Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/dbd89ac4 Branch: refs/heads/spark Commit: dbd89ac4e001f64ef22aa0f955536ac28086b4e8 Parents: 29af41c Author: vikram <vik...@hortonworks.com> Authored: Sat Jun 6 10:46:54 2015 -0700 Committer: vikram <vik...@hortonworks.com> Committed: Sat Jun 6 10:46:54 2015 -0700 ---------------------------------------------------------------------- .../hadoop/hive/ql/exec/tez/TezJobMonitor.java | 2 +- .../hive/ql/exec/tez/TezSessionPoolManager.java | 21 ++++++++++++++++++-- .../hive/ql/exec/tez/TezSessionState.java | 10 ---------- 3 files changed, 20 insertions(+), 13 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hive/blob/dbd89ac4/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezJobMonitor.java ---------------------------------------------------------------------- diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezJobMonitor.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezJobMonitor.java index aa84850..1a4decf 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezJobMonitor.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezJobMonitor.java @@ -127,7 +127,7 @@ public class TezJobMonitor { TezJobMonitor.killRunningJobs(); } try { - for (TezSessionState s: TezSessionState.getOpenSessions()) { + for (TezSessionState s : TezSessionPoolManager.getInstance().getOpenSessions()) { System.err.println("Shutting down tez session."); TezSessionPoolManager.getInstance().close(s, false); } http://git-wip-us.apache.org/repos/asf/hive/blob/dbd89ac4/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezSessionPoolManager.java ---------------------------------------------------------------------- diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezSessionPoolManager.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezSessionPoolManager.java index 1798201..dfa539f 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezSessionPoolManager.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezSessionPoolManager.java @@ -20,13 +20,16 @@ package org.apache.hadoop.hive.ql.exec.tez; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.BlockingQueue; +import java.util.Collections; +import java.util.Iterator; +import java.util.LinkedList; +import java.util.List; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.hadoop.hive.conf.HiveConf; import org.apache.hadoop.hive.ql.metadata.HiveException; import org.apache.hadoop.hive.ql.session.SessionState; -import org.apache.hadoop.hive.shims.ShimLoader; import org.apache.hadoop.hive.shims.Utils; import org.apache.hadoop.security.UserGroupInformation; @@ -49,6 +52,9 @@ public class TezSessionPoolManager { private static TezSessionPoolManager sessionPool = null; + private static List<TezSessionState> openSessions = Collections + .synchronizedList(new LinkedList<TezSessionState>()); + public static TezSessionPoolManager getInstance() throws Exception { if (sessionPool == null) { @@ -68,6 +74,7 @@ public class TezSessionPoolManager { TezSessionState sessionState = defaultQueuePool.take(); newConf.set("tez.queue.name", sessionState.getQueueName()); sessionState.open(newConf); + openSessions.add(sessionState); defaultQueuePool.put(sessionState); } } @@ -148,6 +155,7 @@ public class TezSessionPoolManager { String what = "Created"; if (doOpen) { retTezSessionState.open(conf); + openSessions.add(retTezSessionState); what = "Started"; } @@ -175,6 +183,7 @@ public class TezSessionPoolManager { LOG.info("Closing tez session default? " + tezSessionState.isDefault()); if (!tezSessionState.isDefault()) { tezSessionState.close(keepTmpDir); + openSessions.remove(tezSessionState); } } @@ -184,9 +193,12 @@ public class TezSessionPoolManager { } // we can just stop all the sessions - for (TezSessionState sessionState: TezSessionState.getOpenSessions()) { + Iterator<TezSessionState> iter = openSessions.iterator(); + while (iter.hasNext()) { + TezSessionState sessionState = iter.next(); if (sessionState.isDefault()) { sessionState.close(false); + iter.remove(); } } } @@ -282,5 +294,10 @@ public class TezSessionPoolManager { } close(sessionState, keepTmpDir); sessionState.open(conf, additionalFiles); + openSessions.add(sessionState); + } + + public List<TezSessionState> getOpenSessions() { + return openSessions; } } http://git-wip-us.apache.org/repos/asf/hive/blob/dbd89ac4/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 cef3303..8555c6a 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 @@ -79,9 +79,6 @@ public class TezSessionState { private final Set<LocalResource> localizedResources = new HashSet<LocalResource>(); private boolean doAsEnabled; - private static List<TezSessionState> openSessions - = Collections.synchronizedList(new LinkedList<TezSessionState>()); - /** * Constructor. We do not automatically connect, because we only want to * load tez classes when the user has tez installed. @@ -110,10 +107,6 @@ public class TezSessionState { * Get all open sessions. Only used to clean up at shutdown. * @return List<TezSessionState> */ - public static List<TezSessionState> getOpenSessions() { - return openSessions; - } - public static String makeSessionId() { return UUID.randomUUID().toString(); } @@ -213,8 +206,6 @@ public class TezSessionState { } catch(InterruptedException ie) { //ignore } - - openSessions.add(this); } public void refreshLocalResourcesFromConf(HiveConf conf) @@ -264,7 +255,6 @@ public class TezSessionState { LOG.info("Closing Tez Session"); try { session.stop(); - openSessions.remove(this); } catch (SessionNotRunning nr) { // ignore }
