public IMetaStoreClient get(final HiveConf hiveConf) throws MetaException, IOException, LoginException {
    final HiveClientCacheKey cacheKey = HiveClientCacheKey.fromHiveConf(hiveConf, getThreadId());
    ICacheableMetaStoreClient cacheableHiveMetaStoreClient = null;

    // the hmsc is not shared across threads. So the only way it could get closed while we are doing healthcheck
    // is if removalListener closes it. The synchronization takes care that removalListener won't do it
    synchronized (CACHE_TEARDOWN_LOCK) {
      cacheableHiveMetaStoreClient = getOrCreate(cacheKey);
      cacheableHiveMetaStoreClient.acquire();
    }
    if (!cacheableHiveMetaStoreClient.isOpen()) {
      synchronized (CACHE_TEARDOWN_LOCK) {
        hiveCache.invalidate(cacheKey);
        cacheableHiveMetaStoreClient.close();
        cacheableHiveMetaStoreClient = getOrCreate(cacheKey);
        cacheableHiveMetaStoreClient.acquire();
      }
    }
    return cacheableHiveMetaStoreClient;
  }
@Test
  public void testMultipleThreadAccess() throws ExecutionException, InterruptedException {
    final HiveClientCache cache = new HiveClientCache(1000);

    class GetHiveClient implements Callable<IMetaStoreClient> {
      @Override
      public IMetaStoreClient call() throws IOException, MetaException, LoginException {
        return cache.get(hiveConf);
      }
    }

    ExecutorService executor = Executors.newFixedThreadPool(2);

    Callable<IMetaStoreClient> worker1 = new GetHiveClient();
    Callable<IMetaStoreClient> worker2 = new GetHiveClient();
    Future<IMetaStoreClient> clientFuture1 = executor.submit(worker1);
    Future<IMetaStoreClient> clientFuture2 = executor.submit(worker2);
    IMetaStoreClient client1 = clientFuture1.get();
    IMetaStoreClient client2 = clientFuture2.get();
    assertNotNull(client1);
    assertNotNull(client2);
    assertNotSame(client1, client2);
  }