Hive Metastore ObjectStore PersistenceManager自动关闭的bug

最近在测试HCatalog，由于Hcatalog本身就是一个独立JAR包，虽然它也可以运行service，但是其实这个service就是metastore thrift server，我们在写基于Hcatalog的mapreduce job时候只要把hcatalog JAR包和对应的hive-site.xml文件加入libjars和HADOOP_CLASSPATH中就可以了。不过在测试的时候还是遇到了一些问题，hive metastore server在运行了一段时间后会抛如下错误

2013-06-19 10:35:51,718 ERROR server.TThreadPoolServer (TThreadPoolServer.java:run(182)) - Error occurred during processing of message.
javax.jdo.JDOFatalUserException: Persistence Manager has been closed
        at org.datanucleus.jdo.JDOPersistenceManager.assertIsOpen(JDOPersistenceManager.java:2124)
        at org.datanucleus.jdo.JDOPersistenceManager.currentTransaction(JDOPersistenceManager.java:315)
        at org.apache.hadoop.hive.metastore.ObjectStore.openTransaction(ObjectStore.java:294)
        at org.apache.hadoop.hive.metastore.ObjectStore.getTable(ObjectStore.java:732)
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39)
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
        at java.lang.reflect.Method.invoke(Method.java:597)
        at org.apache.hadoop.hive.metastore.RetryingRawStore.invoke(RetryingRawStore.java:111)
        at com.sun.proxy.$Proxy5.getTable(Unknown Source)
        at org.apache.hadoop.hive.metastore.HiveMetaStore$HMSHandler.get_table(HiveMetaStore.java:982)
        at org.apache.hadoop.hive.metastore.api.ThriftHiveMetastore$Processor$get_table.getResult(ThriftHiveMetastore.java:5017)
        at org.apache.hadoop.hive.metastore.api.ThriftHiveMetastore$Processor$get_table.getResult(ThriftHiveMetastore.java:5005)
        at org.apache.thrift.ProcessFunction.process(ProcessFunction.java:32)
        at org.apache.thrift.TBaseProcessor.process(TBaseProcessor.java:34)

其中PersistenceManager负责控制一组持久化对象包括创建持久化对象和查询对象，它是ObjectStore的一个实例变量，每个ObjectStore拥有一个pm，RawStore是metastore逻辑层和物理底层元数据库（比如derby）交互的接口类，ObjectStore是RawStore的默认实现类。Hive Metastore Server启动的时候会指定一个TProcessor，包装了一个HMSHandler，内部有一个ThreadLocal<RawStore> threadLocalMS实例变量，每个thread维护一个RawStore

private final ThreadLocal<RawStore> threadLocalMS =
  new ThreadLocal<RawStore>() {
    @Override
    protected synchronized RawStore initialValue() {
      return null;
    }
  };

每一个从hive metastore client过来的请求都会从线程池中分配一个WorkerProcess来处理，在HMSHandler中每一个方法都会通过getMS()获取rawstore instance来做具体操作

public RawStore getMS() throws MetaException {
  RawStore ms = threadLocalMS.get();
  if (ms == null) {
    ms = newRawStore();
    threadLocalMS.set(ms);
    ms = threadLocalMS.get();
  }
  return ms;
}

看得出来RawStore是延迟加载，初始化后绑定到threadlocal变量中可以为以后复用

private RawStore newRawStore() throws MetaException {
  LOG.info(addPrefix("Opening raw store with implemenation class:"
      + rawStoreClassName));
  Configuration conf = getConf();  

  return RetryingRawStore.getProxy(hiveConf, conf, rawStoreClassName, threadLocalId.get());
}

RawStore使用了动态代理模式(继承InvocationHandler接口)，内部实现了invoke函数，通过method.invoke()执行真正的逻辑，这样的好处是可以在method.invoke()上下文中添加自己其他的逻辑，RetryingRawStore就是在通过捕捉invoke函数抛出的异常，来达到重试的效果。由于使用reflection机制，异常是wrap在InvocationTargetException中的，不过在hive 0.9中竟然在捕捉到此异常后直接throw出来了，而不是retry，明显不对啊。我对它修改了下，拿出wrap的target exception，判断是不是instance of jdoexception的，再做相应的处理

@Override
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
  Object ret = null;  

  boolean gotNewConnectUrl = false;
  boolean reloadConf = HiveConf.getBoolVar(hiveConf,
      HiveConf.ConfVars.METASTOREFORCERELOADCONF);
  boolean reloadConfOnJdoException = false;  

  if (reloadConf) {
    updateConnectionURL(getConf(), null);
  }  

  int retryCount = 0;
  Exception caughtException = null;
  while (true) {
    try {
      if (reloadConf || gotNewConnectUrl || reloadConfOnJdoException) {
        initMS();
      }
      ret = method.invoke(base, args);
      break;
    } catch (javax.jdo.JDOException e) {
      caughtException = (javax.jdo.JDOException) e.getCause();
    } catch (UndeclaredThrowableException e) {
      throw e.getCause();
    } catch (InvocationTargetException e) {
      Throwable t = e.getTargetException();
      if (t instanceof JDOException){
        caughtException = (JDOException) e.getTargetException();
        reloadConfOnJdoException = true;
        LOG.error("rawstore jdoexception:" + caughtException.toString());
      }else {
          throw e.getCause();
      }
    }  

    if (retryCount >= retryLimit) {
      throw caughtException;
    }  

    assert (retryInterval >= 0);
    retryCount++;
    LOG.error(
        String.format(
            "JDO datastore error. Retrying metastore command " +
                "after %d ms (attempt %d of %d)", retryInterval, retryCount, retryLimit));
    Thread.sleep(retryInterval);
    // If we have a connection error, the JDO connection URL hook might
    // provide us with a new URL to access the datastore.
    String lastUrl = getConnectionURL(getConf());
    gotNewConnectUrl = updateConnectionURL(getConf(), lastUrl);
  }
  return ret;
}

以上是小编为您精心准备的的内容，在的博客、问答、公众号、人物、课程等栏目也有的相关内容，欢迎继续使用右上角搜索按钮进行搜索java
， apache
， hive
， metastore
， threadlocal
， invoke
一个
hive.metastore.uris、hive metastore、hive metastore 启动、hive.metastore.local、hive metastore作用，以便于您获取更多的相关知识。