Repository: hive Updated Branches: refs/heads/master b00621da5 -> 4625964b5
HIVE-15731. sessions are not returned to the sessionPool in case of an interrupt. (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/4625964b Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/4625964b Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/4625964b Branch: refs/heads/master Commit: 4625964b5350343617e7eab45b24ef75d427c0d4 Parents: b00621d Author: Siddharth Seth <ss...@apache.org> Authored: Wed Jan 25 20:54:03 2017 -0800 Committer: Siddharth Seth <ss...@apache.org> Committed: Wed Jan 25 20:54:03 2017 -0800 ---------------------------------------------------------------------- .../hive/ql/exec/tez/TezSessionPoolManager.java | 44 +++++++++++++------- .../apache/hadoop/hive/ql/exec/tez/TezTask.java | 10 ++++- 2 files changed, 38 insertions(+), 16 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hive/blob/4625964b/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 fae6393..d7148bb 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 @@ -322,23 +322,39 @@ public class TezSessionPoolManager { public void returnSession(TezSessionState tezSessionState, boolean llap) throws Exception { - if (llap && (this.numConcurrentLlapQueries > 0)) { - llapQueue.release(); - } - if (tezSessionState.isDefault() && tezSessionState instanceof TezSessionPoolSession) { - LOG.info("The session " + tezSessionState.getSessionId() - + " belongs to the pool. Put it back in"); - SessionState sessionState = SessionState.get(); - if (sessionState != null) { - sessionState.setTezSession(null); + // Ignore the interrupt status while returning the session, but set it back + // on the thread in case anything else needs to deal with it. + boolean isInterrupted = Thread.interrupted(); + + try { + if (isInterrupted) { + LOG.info("returnSession invoked with interrupt status set"); + } + if (llap && (this.numConcurrentLlapQueries > 0)) { + llapQueue.release(); } - TezSessionPoolSession poolSession = (TezSessionPoolSession)tezSessionState; - if (poolSession.returnAfterUse()) { - defaultQueuePool.put(poolSession); + if (tezSessionState.isDefault() && + tezSessionState instanceof TezSessionPoolSession) { + LOG.info("The session " + tezSessionState.getSessionId() + + " belongs to the pool. Put it back in"); + SessionState sessionState = SessionState.get(); + if (sessionState != null) { + sessionState.setTezSession(null); + } + TezSessionPoolSession poolSession = + (TezSessionPoolSession) tezSessionState; + if (poolSession.returnAfterUse()) { + defaultQueuePool.put(poolSession); + } + } + // non default session nothing changes. The user can continue to use the existing + // session in the SessionState + } finally { + // Reset the interrupt status. + if (isInterrupted) { + Thread.currentThread().interrupt(); } } - // non default session nothing changes. The user can continue to use the existing - // session in the SessionState } public static void closeIfNotDefault( http://git-wip-us.apache.org/repos/asf/hive/blob/4625964b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezTask.java ---------------------------------------------------------------------- diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezTask.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezTask.java index 0efca68..7479b85 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezTask.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezTask.java @@ -190,12 +190,18 @@ public class TezTask extends Task<TezWork> { counters = dagClient.getDAGStatus(statusGetOpts).getDAGCounters(); } catch (Exception err) { // Don't fail execution due to counters - just don't print summary info - LOG.error("Failed to get counters: " + err, err); + LOG.warn("Failed to get counters. Ignoring, summary info will be incomplete. " + err, err); counters = null; } } finally { // We return this to the pool even if it's unusable; reopen is supposed to handle this. - TezSessionPoolManager.getInstance().returnSession(session, getWork().getLlapMode()); + try { + TezSessionPoolManager.getInstance() + .returnSession(session, getWork().getLlapMode()); + } catch (Exception e) { + LOG.error("Failed to return session: {} to pool", session, e); + throw e; + } } if (LOG.isInfoEnabled() && counters != null
