You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by sa...@apache.org on 2018/11/13 10:56:36 UTC
hive git commit: HIVE-20682: Async query execution can potentially
fail if shared sessionHive is closed by master thread (Sankar Hariappan,
reviewed by Mahesh Kumar Behera, Anishek Agarwal)
Repository: hive
Updated Branches:
refs/heads/master af4017028 -> 99d25f024
HIVE-20682: Async query execution can potentially fail if shared sessionHive is closed by master thread (Sankar Hariappan, reviewed by Mahesh Kumar Behera, Anishek Agarwal)
Project: http://git-wip-us.apache.org/repos/asf/hive/repo
Commit: http://git-wip-us.apache.org/repos/asf/hive/commit/99d25f02
Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/99d25f02
Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/99d25f02
Branch: refs/heads/master
Commit: 99d25f02421a84bf0f96660f9248fd6518dc7c8a
Parents: af40170
Author: Sankar Hariappan <sa...@apache.org>
Authored: Tue Nov 13 16:26:04 2018 +0530
Committer: Sankar Hariappan <sa...@apache.org>
Committed: Tue Nov 13 16:26:04 2018 +0530
----------------------------------------------------------------------
.../hive/ql/parse/TestReplicationScenarios.java | 2 +-
.../apache/hadoop/hive/ql/metadata/Hive.java | 129 +++++++++++--------
.../service/cli/operation/SQLOperation.java | 20 +--
.../service/cli/session/HiveSessionImpl.java | 56 +++++++-
.../cli/session/TestSessionManagerMetrics.java | 2 +
5 files changed, 138 insertions(+), 71 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hive/blob/99d25f02/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/parse/TestReplicationScenarios.java
----------------------------------------------------------------------
diff --git a/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/parse/TestReplicationScenarios.java b/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/parse/TestReplicationScenarios.java
index 75cd68a..5a88550 100644
--- a/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/parse/TestReplicationScenarios.java
+++ b/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/parse/TestReplicationScenarios.java
@@ -389,7 +389,7 @@ public class TestReplicationScenarios {
Task replLoadTask = TaskFactory.get(replLoadWork, confTemp);
replLoadTask.initialize(null, null, new DriverContext(driver.getContext()), null);
replLoadTask.executeTask(null);
- Hive.getThreadLocal().closeCurrent();
+ Hive.closeCurrent();
return replLoadWork.getRootTask();
}
http://git-wip-us.apache.org/repos/asf/hive/blob/99d25f02/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java b/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java
index 180b41e..e185bf4 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java
@@ -167,35 +167,36 @@ public class Hive {
private IMetaStoreClient metaStoreClient;
private SynchronizedMetaStoreClient syncMetaStoreClient;
private UserGroupInformation owner;
+ private boolean isAllowClose = true;
// metastore calls timing information
private final ConcurrentHashMap<String, Long> metaCallTimeMap = new ConcurrentHashMap<>();
- // Static class to store thread local Hive object and allowClose flag.
+ // Static class to store thread local Hive object.
private static class ThreadLocalHive extends ThreadLocal<Hive> {
- private ThreadLocal<Boolean> allowClose = ThreadLocal.withInitial(() -> true);
-
@Override
protected Hive initialValue() {
return null;
}
@Override
- public synchronized void remove() {
- if (allowClose() && (this.get() != null)) {
- this.get().close();
+ public synchronized void set(Hive hiveObj) {
+ Hive currentHive = this.get();
+ if (currentHive != hiveObj) {
+ // Remove/close current thread-local Hive object before overwriting with new Hive object.
+ remove();
+ super.set(hiveObj);
}
- super.remove();
- this.allowClose.set(true);
- }
-
- public synchronized void set(Hive hiveObj, boolean allowClose) {
- super.set(hiveObj);
- this.allowClose.set(allowClose);
}
- boolean allowClose() {
- return this.allowClose.get();
+ @Override
+ public synchronized void remove() {
+ Hive currentHive = this.get();
+ if (currentHive != null) {
+ // Close the metastore connections before removing it from thread local hiveDB.
+ currentHive.close(false);
+ super.remove();
+ }
}
}
@@ -317,7 +318,12 @@ public class Hive {
Hive db = hiveDB.get();
if (db == null || !db.isCurrentUserOwner() || needsRefresh
|| (c != null && !isCompatible(db, c, isFastCheck))) {
- db = create(c, false, db, doRegisterAllFns);
+ if (db != null) {
+ LOG.debug("Creating new db. db = " + db + ", needsRefresh = " + needsRefresh +
+ ", db.isCurrentUserOwner = " + db.isCurrentUserOwner());
+ closeCurrent();
+ }
+ db = create(c, doRegisterAllFns);
}
if (c != null) {
db.conf = c;
@@ -325,26 +331,16 @@ public class Hive {
return db;
}
- private static Hive create(HiveConf c, boolean needsRefresh, Hive db, boolean doRegisterAllFns)
- throws HiveException {
- if (db != null) {
- LOG.debug("Creating new db. db = " + db + ", needsRefresh = " + needsRefresh +
- ", db.isCurrentUserOwner = " + db.isCurrentUserOwner());
- if (hiveDB.allowClose()) {
- db.close();
- }
- }
- closeCurrent();
+ private static Hive create(HiveConf c, boolean doRegisterAllFns) throws HiveException {
if (c == null) {
c = createHiveConf();
}
c.set("fs.scheme.class", "dfs");
Hive newdb = new Hive(c, doRegisterAllFns);
- hiveDB.set(newdb, true);
+ hiveDB.set(newdb);
return newdb;
}
-
private static HiveConf createHiveConf() {
SessionState session = SessionState.get();
return (session == null) ? new HiveConf(Hive.class) : session.getConf();
@@ -360,6 +356,18 @@ public class Hive {
}
}
+ private boolean isCurrentUserOwner() throws HiveException {
+ try {
+ return owner == null || owner.equals(UserGroupInformation.getCurrentUser());
+ } catch(IOException e) {
+ throw new HiveException("Error getting current user: " + e.getMessage(), e);
+ }
+ }
+
+ public static Hive getThreadLocal() {
+ return hiveDB.get();
+ }
+
public static Hive get() throws HiveException {
return get(true);
}
@@ -383,21 +391,13 @@ public class Hive {
}
public static void set(Hive hive) {
- hiveDB.set(hive, true);
- }
-
- public static void set(Hive hive, boolean allowClose) {
- hiveDB.set(hive, allowClose);
+ hiveDB.set(hive);
}
public static void closeCurrent() {
hiveDB.remove();
}
- public static Hive getThreadLocal() {
- return hiveDB.get();
- }
-
/**
* Hive
*
@@ -411,30 +411,49 @@ public class Hive {
}
}
-
- private boolean isCurrentUserOwner() throws HiveException {
- try {
- return owner == null || owner.equals(UserGroupInformation.getCurrentUser());
- } catch(IOException e) {
- throw new HiveException("Error getting current user: " + e.getMessage(), e);
- }
+ /**
+ * GC is attempting to destroy the object.
+ * No one references this Hive anymore, so HMS connection from this Hive object can be closed.
+ * @throws Throwable
+ */
+ @Override
+ protected void finalize() throws Throwable {
+ close(true);
+ super.finalize();
}
+ /**
+ * Marks if the given Hive object is allowed to close metastore connections.
+ * @param allowClose
+ */
+ public void setAllowClose(boolean allowClose) {
+ isAllowClose = allowClose;
+ }
+ /**
+ * Gets the allowClose flag which determines if it is allowed to close metastore connections.
+ * @returns allowClose flag
+ */
+ public boolean allowClose() {
+ return isAllowClose;
+ }
/**
- * closes the connection to metastore for the calling thread
+ * Closes the connection to metastore for the calling thread if allow to close.
+ * @param forceClose - Override the isAllowClose flag to forcefully close the MS connections.
*/
- private void close() {
- LOG.debug("Closing current thread's connection to Hive Metastore.");
- if (metaStoreClient != null) {
- metaStoreClient.close();
- metaStoreClient = null;
- }
- // syncMetaStoreClient is wrapped on metaStoreClient. So, it is enough to close it once.
- syncMetaStoreClient = null;
- if (owner != null) {
- owner = null;
+ public void close(boolean forceClose) {
+ if (allowClose() || forceClose) {
+ LOG.debug("Closing current thread's connection to Hive Metastore.");
+ if (metaStoreClient != null) {
+ metaStoreClient.close();
+ metaStoreClient = null;
+ }
+ // syncMetaStoreClient is wrapped on metaStoreClient. So, it is enough to close it once.
+ syncMetaStoreClient = null;
+ if (owner != null) {
+ owner = null;
+ }
}
}
http://git-wip-us.apache.org/repos/asf/hive/blob/99d25f02/service/src/java/org/apache/hive/service/cli/operation/SQLOperation.java
----------------------------------------------------------------------
diff --git a/service/src/java/org/apache/hive/service/cli/operation/SQLOperation.java b/service/src/java/org/apache/hive/service/cli/operation/SQLOperation.java
index 3d24884..f975199 100644
--- a/service/src/java/org/apache/hive/service/cli/operation/SQLOperation.java
+++ b/service/src/java/org/apache/hive/service/cli/operation/SQLOperation.java
@@ -299,6 +299,11 @@ public class SQLOperation extends ExecuteStatementOperation {
private BackgroundWork(UserGroupInformation currentUGI,
Hive parentHive,
SessionState parentSessionState, boolean asyncPrepare) {
+ // Note: parentHive can be shared by multiple threads and so it should be protected from any
+ // thread closing metastore connections when some other thread still accessing it. So, it is
+ // expected that allowClose flag in parentHive is set to false by caller and it will be caller's
+ // responsibility to close it explicitly with forceClose flag as true.
+ // Shall refer to sessionHive in HiveSessionImpl.java for the usage.
this.currentUGI = currentUGI;
this.parentHive = parentHive;
this.parentSessionState = parentSessionState;
@@ -310,7 +315,8 @@ public class SQLOperation extends ExecuteStatementOperation {
PrivilegedExceptionAction<Object> doAsAction = new PrivilegedExceptionAction<Object>() {
@Override
public Object run() throws HiveSQLException {
- Hive.set(parentHive, false);
+ assert (!parentHive.allowClose());
+ Hive.set(parentHive);
// TODO: can this result in cross-thread reuse of session state?
SessionState.setCurrentSessionState(parentSessionState);
PerfLogger.setPerfLogger(SessionState.getPerfLogger());
@@ -328,13 +334,11 @@ public class SQLOperation extends ExecuteStatementOperation {
LOG.error("Error running hive query: ", e);
} finally {
LogUtils.unregisterLoggingContext();
- Hive hiveDb = Hive.getThreadLocal();
- if (hiveDb != null && hiveDb != parentHive) {
- // If new hive object is created by the child thread, then we need to close it as it might
- // have created a hms connection. Call Hive.closeCurrent() that closes the HMS connection, causes
- // HMS connection leaks otherwise.
- Hive.closeCurrent();
- }
+
+ // If new hive object is created by the child thread, then we need to close it as it might
+ // have created a hms connection. Call Hive.closeCurrent() that closes the HMS connection, causes
+ // HMS connection leaks otherwise.
+ Hive.closeCurrent();
}
return null;
}
http://git-wip-us.apache.org/repos/asf/hive/blob/99d25f02/service/src/java/org/apache/hive/service/cli/session/HiveSessionImpl.java
----------------------------------------------------------------------
diff --git a/service/src/java/org/apache/hive/service/cli/session/HiveSessionImpl.java b/service/src/java/org/apache/hive/service/cli/session/HiveSessionImpl.java
index 0018f68..a8bf876 100644
--- a/service/src/java/org/apache/hive/service/cli/session/HiveSessionImpl.java
+++ b/service/src/java/org/apache/hive/service/cli/session/HiveSessionImpl.java
@@ -185,11 +185,10 @@ public class HiveSessionImpl implements HiveSession {
LOG.error(msg, e);
throw new HiveSQLException(msg, e);
}
- try {
- sessionHive = Hive.get(getHiveConf());
- } catch (HiveException e) {
- throw new HiveSQLException("Failed to get metastore connection", e);
- }
+
+ // Set sessionHive object created based on sessionConf.
+ setSessionHive();
+
// Process global init file: .hiverc
processGlobalInitFile();
// Set fetch size in session conf map
@@ -237,6 +236,28 @@ public class HiveSessionImpl implements HiveSession {
}
}
+ /**
+ * Sets sessionHive object created based on sessionConf.
+ * @throws HiveSQLException
+ */
+ private void setSessionHive() throws HiveSQLException {
+ Hive newSessionHive;
+ try {
+ newSessionHive = Hive.get(getHiveConf());
+
+ // HMS connections from sessionHive shouldn't be closed by any query execution thread when it
+ // recreates the Hive object. It is allowed to be closed only when session is closed/released.
+ newSessionHive.setAllowClose(false);
+ } catch (HiveException e) {
+ throw new HiveSQLException("Failed to get metastore connection", e);
+ }
+
+ // The previous sessionHive object might still be referred by any async query execution thread.
+ // So, it shouldn't be closed here explicitly. Anyways, Hive object will auto-close HMS connection
+ // when it is garbage collected. So, it is safe to just overwrite sessionHive here.
+ sessionHive = newSessionHive;
+ }
+
private void processGlobalInitFile() {
IHiveFileProcessor processor = new GlobalHivercFileProcessor();
@@ -402,7 +423,20 @@ public class HiveSessionImpl implements HiveSession {
}
// set the thread name with the logging prefix.
sessionState.updateThreadName();
- Hive.set(sessionHive);
+
+ // If the thread local Hive is different from sessionHive, it means, the previous query execution in
+ // master thread has re-created Hive object due to changes in MS related configurations in sessionConf.
+ // So, it is necessary to reset sessionHive object based on new sessionConf. Here, we cannot,
+ // directly set sessionHive with thread local Hive because if the previous command was REPL LOAD, then
+ // the config changes lives only within command execution not in session level.
+ // So, the safer option is to invoke Hive.get() which decides if to reuse Thread local Hive or re-create it.
+ if (Hive.getThreadLocal() != sessionHive) {
+ try {
+ setSessionHive();
+ } catch (HiveSQLException e) {
+ throw new RuntimeException(e);
+ }
+ }
}
/**
@@ -777,12 +811,20 @@ public class HiveSessionImpl implements HiveSession {
}
if (sessionHive != null) {
try {
- Hive.closeCurrent();
+ sessionHive.close(true);
} catch (Throwable t) {
LOG.warn("Error closing sessionHive", t);
}
sessionHive = null;
}
+ try {
+ // The thread local Hive in master thread can be different from sessionHive if any query
+ // execution from master thread resets it to new Hive object due to changes in sessionConf.
+ // So, need to close it as well. If it is same as sessionHive, then it is just no-op.
+ Hive.closeCurrent();
+ } catch (Throwable t) {
+ LOG.warn("Error closing thread local Hive", t);
+ }
release(true, false);
}
}
http://git-wip-us.apache.org/repos/asf/hive/blob/99d25f02/service/src/test/org/apache/hive/service/cli/session/TestSessionManagerMetrics.java
----------------------------------------------------------------------
diff --git a/service/src/test/org/apache/hive/service/cli/session/TestSessionManagerMetrics.java b/service/src/test/org/apache/hive/service/cli/session/TestSessionManagerMetrics.java
index 5655458..be8d70b 100644
--- a/service/src/test/org/apache/hive/service/cli/session/TestSessionManagerMetrics.java
+++ b/service/src/test/org/apache/hive/service/cli/session/TestSessionManagerMetrics.java
@@ -280,6 +280,7 @@ public class TestSessionManagerMetrics {
@Override
public void run() {
try {
+ Hive.set(session.getSessionHive());
OperationHandle handle = session.getTables("catalog", "schema", "table", null);
session.closeOperation(handle);
} catch (Exception e) {
@@ -334,6 +335,7 @@ public class TestSessionManagerMetrics {
@Override
public void run() {
try {
+ Hive.set(session.getSessionHive());
OperationHandle handle = session.getTables("catalog", "schema", "table", null);
session.closeOperation(handle);
} catch (Exception e) {