问题现象
生产环境hiveserver2使用过程中占用大量内存，甚至将内存上限增大到32G左右后hiveserver2仍会达到内存上限。使用G1GC，单次full GC耗时很长，且绝大部分内存无法被回收（只能回收几百M内存），然后发生OOM退出。怀疑存在内存泄漏问题。本文围绕hiveserver2内存泄漏问题展开分析。
生产环境Hive版本为3.1.0。
Heap dump 分析
在生产服务器获取到hiveserver2 OOM时候的heap dump之后，使用MAT工具分析。获取它的leak suspect报告。具体分析步骤参见Java Heap Dump 分析步骤。
我们使用浏览器打开leak suspect报告。发现里面有一个内存泄漏怀疑点。打开详情，内容如下图所示：
上图中可以很明显的看出operationManager中有一个叫做queryIdOperation的ConcurrentHashMap，占用了大量的内存。这个报告给出了问题分析的方向。下面的分析围绕着queryIdOpereation展开。
原因分析
我们查看源代码，发现OperationManager类中的queryIdOperation为私有变量。因此queryIdOperation只可能在OperationManager中操作。继续寻找操作queryIdOperation的方法，发现有如下三个：
- addOperation
- removeOperation
- getOperationByQueryId
其中前两个方法分别为向集合中添加和移除元素。我们接下来分析这两个方法。
OperationManager的addOperation方法代码如下：
private void addOperation(Operation operation) {
LOG.info("Adding operation: " + operation.getHandle());
// 通过getQueryId方法从operation中获取queryId，然后将queryId作为key，存放入queryIdOperation
queryIdOperation.put(getQueryId(operation), operation);
handleToOperation.put(operation.getHandle(), operation);
if (operation instanceof SQLOperation) {
synchronized (webuiLock) {
liveQueryInfos.put(operation.getHandle().getHandleIdentifier().toString(),
((SQLOperation) operation).getQueryInfo());
}
}
}
getQueryId方法内容如下：
private String getQueryId(Operation operation) {
// 获取parent session的HiveConf对象
// queryId在这个HiveConf对象当中存放
return operation.getParentSession().getHiveConf().getVar(ConfVars.HIVEQUERYID);
}
这个方法是重点，此处先埋个伏笔，后面还会再次分析这个方法。
removeOperation方法逻辑如下：
private Operation removeOperation(OperationHandle opHandle) {
Operation operation = handleToOperation.remove(opHandle);
// 通过上面的逻辑，获取queryId
String queryId = getQueryId(operation);
// 从queryIdOperation集合中remove掉
queryIdOperation.remove(queryId);
LOG.info("Removed queryId: {} corresponding to operation: {}", queryId, opHandle);
if (operation instanceof SQLOperation) {
removeSafeQueryInfo(opHandle);
}
return operation;
}
接下来我们需要顺着向上层找，分别追踪addOperation和removeOperation方法的调用链。
addOperation方法在OperationManager的newExecuteStatementOperation方法中调用，内容如下：
public ExecuteStatementOperation newExecuteStatementOperation(HiveSession parentSession,
String statement, Map<String, String> confOverlay, boolean runAsync, long queryTimeout)
throws HiveSQLException {
// 创建一个ExecuteStatementOperation
ExecuteStatementOperation executeStatementOperation =
ExecuteStatementOperation.newExecuteStatementOperation(parentSession, statement,
confOverlay, runAsync, queryTimeout);
// 调用addOperation
addOperation(executeStatementOperation);
return executeStatementOperation;
}
追溯这个方法调用，我们来到HiveSessionImpl的executeStatementInternal方法，内容如下：
private OperationHandle executeStatementInternal(String statement,
Map<String, String> confOverlay, boolean runAsync, long queryTimeout) throws HiveSQLException {
acquire(true, true);
ExecuteStatementOperation operation = null;
OperationHandle opHandle = null;
try {
// 此处调用了newExecuteStatementOperation
operation = getOperationManager().newExecuteStatementOperation(getSession(), statement,
confOverlay, runAsync, queryTimeout);
opHandle = operation.getHandle();
addOpHandle(opHandle);
operation.run();
return opHandle;
} catch (HiveSQLException e) {
// Refering to SQLOperation.java, there is no chance that a HiveSQLException throws and the
// async background operation submits to thread pool successfully at the same time. So, Cleanup
// opHandle directly when got HiveSQLException
if (opHandle != null) {
removeOpHandle(opHandle);
getOperationManager().closeOperation(opHandle);
}
throw e;
} finally {
if (operation == null || operation.getBackgroundHandle() == null) {
release(true, true); // Not async, or wasn't submitted for some reason (failure, etc.)
} else {
releaseBeforeOpLock(true); // Release, but keep the lock (if present).
}
}
}
再继续向上追踪，发现上面的方法在HiveSessionImpl的executeStatement和executeStatementAsync方法中调用（忽略了重载方法）。这两个方法分别为阻塞方式执行SQL statement和异步执行SQL statement。如果继续向上追踪调用，我们能够找到CLIService类。ThriftCLIService类又再次包装了CLIService类，它拥有ExecuteStatement方法。这个方法是thrift RPC调用的endpoint，通过TExecuteStatementReq类传递调用参数。继续追踪调用端的话我们陆续跟踪到HiveStatement的runAsyncOnServer方法->execute方法（具体逻辑不再分析，只分析调用链）。HiveStatement是java.sql.Statement的子类，因此再往上分析就是JDBC使用的范畴了。调用链分析到这里为止。我们得到的结论是Hive JDBC调用statement的execute方法，会在执行SQL前创建一个operation。一条SQL的执行对应着一个operation。
接下来我们转到RemoveOperation的分析。它的调用位于HiveSessionImpl的close方法。close方法会关闭所有的operation。代码如下所示，其他不相关的逻辑此处不分析。
@Override
public void close() throws HiveSQLException {
try {
acquire(true, false);
// Iterate through the opHandles and close their operations
List<OperationHandle> ops = null;
synchronized (opHandleSet) {
ops = new ArrayList<>(opHandleSet);
opHandleSet.clear();
}
// 遍历各个operationHandle，一个operationHandle对应着一个operation
// 然后关闭他们
for (OperationHandle opHandle : ops) {
operationManager.closeOperation(opHandle);
}
// Cleanup session log directory.
cleanupSessionLogDir();
HiveHistory hiveHist = sessionState.getHiveHistory();
if (null != hiveHist) {
hiveHist.closeStream();
}
try {
sessionState.resetThreadName();
sessionState.close();
} finally {
sessionState = null;
}
} catch (IOException ioe) {
throw new HiveSQLException("Failure to close", ioe);
} finally {
if (sessionState != null) {
try {
sessionState.resetThreadName();
sessionState.close();
} catch (Throwable t) {
LOG.warn("Error closing session", t);
}
sessionState = null;
}
if (sessionHive != null) {
try {
Hive.closeCurrent();
} catch (Throwable t) {
LOG.warn("Error closing sessionHive", t);
}
sessionHive = null;
}
release(true, false);
}
}
除此之外还有一处调用位于closeOperation方法，内容如下。
@Override
public void closeOperation(OperationHandle opHandle) throws HiveSQLException {
acquire(true, false);
try {
operationManager.closeOperation(opHandle);
synchronized (opHandleSet) {
opHandleSet.remove(opHandle);
}
} finally {
release(true, false);
}
}
继续追踪调用，我们发现它的调用端也在CLIService的closeOperation方法。继续跟踪到ThriftCLIService的CloseOperation方法，它也是thrift RPC endpoint。通过TCloseOperationReq传递RPC调用参数。追踪到RPC调用端，我们跟踪到HiveStatement的closeStatementIfNeeded方法。在往上追踪，调用链为closeClientOperation -> close方法。其中close方法重写了java.sql.Statement的同名方法。到这里我们得到结论，close Hive的statement的时候，会调用removeOperation，从而将operation从queryIdOperation中移除。
按照JDBC标准使用方式，statement使用完毕之后是必须要close的。也就是说正常情况下addOperation和removeOperation必然是成对出现。我们先假设用户使用问题，没有及时close掉statement。
接着继续分析还有哪些时机会调用removeOperation方法。我们找到HiveSessionImpl的close方法 -> SessionManager的closeSession方法。除了正常关闭session外，SessionManager中还有一个startTimeoutChecker。这个方法周期运行，当session超时的时候会自动关闭session。从而关闭所有的statement。这些措施确保了removeOperation是一定会被调用到的。就算是用户使用问题，没有close掉statement，这些operation也是可以被清理掉的。
造成OOM的原因是某些operation始终不能够被remove掉。查看日志我们的确发现部分query id的确没有被remove掉(removeOperation中LOG.info("Removed queryId: {} corresponding to operation: {}", queryId, opHandle);这一行代码会打印日志，存在一些query id没有这一行日志)。问题可能在于OperationManager的getQueryId方法。无法通过operation获取到它对应的query id。
我们回到OperationManager的getQueryId方法。发现query id并没有存储在operation中，而是存储在HiveConf中。
private String getQueryId(Operation operation) {
return operation.getParentSession().getHiveConf().getVar(ConfVars.HIVEQUERYID);
}
一路跟踪operation的parentSession是什么时候赋值进去的。最终找到了HiveSessionImpl的executeStatementInternal方法。下面只贴出关键的一行，其他无关部分省略。
operation = getOperationManager().newExecuteStatementOperation(getSession(), statement,
confOverlay, runAsync, queryTimeout);
getSession方法返回的是this。说明这些operation共用同一个Hive Session（同一个JDBC连接下所有操作公用session）。自然HiveConf也是公用的。到这里为止分析的重点来到了这个HiveConf保存的内容上。
Hive的query id存储在HiveConf中，key为ConfVars.HIVEQUERYID。猜测这个key一定有某个地方被set。跟踪HiveConf的set这个key的调用，我们发现QueryState的build方法。
QueryState的build方法中分配新的queryId。方法内容如下：
public QueryState build() {
HiveConf queryConf;
if (isolated) {
// isolate query conf
if (hiveConf == null) {
queryConf = new HiveConf();
} else {
queryConf = new HiveConf(hiveConf);
}
} else {
queryConf = hiveConf;
}
// Set the specific parameters if needed
if (confOverlay != null && !confOverlay.isEmpty()) {
// apply overlay query specific settings, if any
for (Map.Entry<String, String> confEntry : confOverlay.entrySet()) {
try {
queryConf.verifyAndSet(confEntry.getKey(), confEntry.getValue());
} catch (IllegalArgumentException e) {
throw new RuntimeException("Error applying statement specific settings", e);
}
}
}
// Generate the new queryId if needed
// 如果需要生成新的query id
if (generateNewQueryId) {
// 分配新的query id
String queryId = QueryPlan.makeQueryId();
queryConf.setVar(HiveConf.ConfVars.HIVEQUERYID, queryId);
// FIXME: druid storage handler relies on query.id to maintain some staging directories
// expose queryid to session level
// 将query id存放到hive session中
if (hiveConf != null) {
hiveConf.setVar(HiveConf.ConfVars.HIVEQUERYID, queryId);
}
}
QueryState queryState = new QueryState(queryConf);
if (lineageState != null) {
queryState.setLineageState(lineageState);
}
return queryState;
}
下面我们要确认下这个build方法是否在执行SQL查询的过程中调用。跟踪调用我们发现Operation类的构造函数。内容如下：
protected Operation(HiveSession parentSession,
Map<String, String> confOverlay, OperationType opType) {
this.parentSession = parentSession;
this.opHandle = new OperationHandle(opType, parentSession.getProtocolVersion());
beginTime = System.currentTimeMillis();
lastAccessTime = beginTime;
operationTimeout = HiveConf.getTimeVar(parentSession.getHiveConf(),
HiveConf.ConfVars.HIVE_SERVER2_IDLE_OPERATION_TIMEOUT, TimeUnit.MILLISECONDS);
scheduledExecutorService = Executors.newScheduledThreadPool(1);
currentStateScope = updateOperationStateMetrics(null, MetricsConstant.OPERATION_PREFIX,
MetricsConstant.COMPLETED_OPERATION_PREFIX, state);
// 这里创建出了queryState
// 这个queryState被operation持有
queryState = new QueryState.Builder()
.withConfOverlay(confOverlay)
// 指定需要生成query id
.withGenerateNewQueryId(true)
.withHiveConf(parentSession.getHiveConf())
.build();
}
跟踪这个构造函数，不难发现ExecuteStatementOperation是Operation的子类。创建ExecuteStatementOperation的时候调用了这个方法。
public ExecuteStatementOperation(HiveSession parentSession, String statement,
Map<String, String> confOverlay, boolean runInBackground) {
super(parentSession, confOverlay, OperationType.EXECUTE_STATEMENT);
this.statement = statement;
}
ExecuteStatementOperation是HiveCommandOperation的父类。HiveCommandOperation的构造函数中自然需要调用上面的方法。
ExecuteStatementOperation还有一个方法newExecuteStatementOperation。这个方法我们上面已经分析过了，它最后创建了一个HiveCommandOperation对象并返回。经过这段分析我们验证了Hive每次执行SQL statement的时候都会设置一个新的query id。那么问题来了，如果上一个query id还被来得及被remove就设置了新的query id，上一个query id就再也没有机会被remove，造成OOM的问题。同一个session只会保存最后一个query id。到此问题的根源已经找到。
问题解决
跟踪社区我们发现在Hive项目的branch-3.1分支中有一个HIVE-26530patch。这个patch合并的时间明显晚于Hive 3.1.0发布的时间，是一个hotfix。它对应的正是OperationManager的getQueryId方法的修改。这个patch将OperationManager的getQueryId方法从：
private String getQueryId(Operation operation) {
return operation.getParentSession().getHiveConf().getVar(ConfVars.HIVEQUERYID);
}
修改为：
private String getQueryId(Operation operation) {
return operation.getQueryId();
}
Operation类增加如下代码：
public String getQueryId() {
return queryState.getQueryId();
}
该patch做出的改动将query id保存在每个operation专有的queryState中，从而杜绝了query id被覆盖的情况。将本地Hive3.1.0代码合入这个patch后重新编译。替换集群中的hive-service-xxx.jar为新编译输出的jar后重启集群，问题解决。目前使用Hive 3.x版本的环境都存在此隐患，建议紧急修复此问题。
本博客为作者原创，欢迎大家参与讨论和批评指正。如需转载请注明出处。