@VisibleForTesting
  static Key extractKey(String cacheKeys, Configuration conf) {
    // generate key elements in a certain order, so that the Key instances are comparable
    List<Object> elements = Lists.newArrayList();
    elements.add(conf.get(HiveConf.ConfVars.METASTORE_URIS.varname, ""));
    elements.add(conf.get(HiveCatalog.HIVE_CONF_CATALOG, "hive"));
    if (cacheKeys == null || cacheKeys.isEmpty()) {
      return Key.of(elements);
    }

    Set<KeyElementType> types = Sets.newTreeSet(Comparator.comparingInt(Enum::ordinal));
    Map<String, String> confElements = Maps.newTreeMap();
    for (String element : cacheKeys.split(",", -1)) {
      String trimmed = element.trim();
      if (trimmed.toLowerCase(Locale.ROOT).startsWith(CONF_ELEMENT_PREFIX)) {
        String key = trimmed.substring(CONF_ELEMENT_PREFIX.length());
        ValidationException.check(
            !confElements.containsKey(key), "Conf key element %s already specified", key);
        confElements.put(key, conf.get(key));
      } else {
        KeyElementType type = KeyElementType.valueOf(trimmed.toUpperCase());
        switch (type) {
          case UGI:
          case USER_NAME:
            ValidationException.check(
                !types.contains(type), "%s key element already specified", type.name());
            types.add(type);
            break;
          default:
            throw new ValidationException("Unknown key element %s", trimmed);
        }
      }
    }
    for (KeyElementType type : types) {
      switch (type) {
        case UGI:
          try {
            elements.add(UserGroupInformation.getCurrentUser());
          } catch (IOException e) {
            throw new UncheckedIOException(e);
          }
          break;
        case USER_NAME:
          try {
            elements.add(UserGroupInformation.getCurrentUser().getUserName());
          } catch (IOException e) {
            throw new UncheckedIOException(e);
          }
          break;
        default:
          throw new RuntimeException("Unexpected key element " + type.name());
      }
    }
    for (String key : confElements.keySet()) {
      elements.add(ConfElement.of(key, confElements.get(key)));
    }
    return Key.of(elements);
  }
@Test
  public void testCacheKey() throws Exception {
    UserGroupInformation current = UserGroupInformation.getCurrentUser();
    UserGroupInformation foo1 = UserGroupInformation.createProxyUser("foo", current);
    UserGroupInformation foo2 = UserGroupInformation.createProxyUser("foo", current);
    UserGroupInformation bar = UserGroupInformation.createProxyUser("bar", current);

    HiveConf hiveConf = HIVE_METASTORE_EXTENSION.hiveConf();

    Key key1 =
        foo1.doAs(
          (PrivilegedAction<Key>)
            () -> CachedClientPool.extractKey("user_name,conf:key1", hiveConf));
    Key key2 =
        foo2.doAs(
          (PrivilegedAction<Key>)
            () -> CachedClientPool.extractKey("conf:key1,user_name", hiveConf));
    assertThat(key2).as("Key elements order shouldn't matter").isEqualTo(key1);

    key1 = foo1.doAs((PrivilegedAction<Key>) () -> CachedClientPool.extractKey("ugi", hiveConf));
    key2 = bar.doAs((PrivilegedAction<Key>) () -> CachedClientPool.extractKey("ugi", hiveConf));
    assertThat(key2).as("Different users are not supposed to be equivalent").isNotEqualTo(key1);

    key2 = foo2.doAs((PrivilegedAction<Key>) () -> CachedClientPool.extractKey("ugi", hiveConf));
    assertThat(key2)
        .as("Different UGI instances are not supposed to be equivalent")
        .isNotEqualTo(key1);

    key1 = CachedClientPool.extractKey("ugi", hiveConf);
    key2 = CachedClientPool.extractKey("ugi,conf:key1", hiveConf);
    assertThat(key2)
        .as("Keys with different number of elements are not supposed to be equivalent")
        .isNotEqualTo(key1);

    Configuration conf1 = new Configuration(hiveConf);
    Configuration conf2 = new Configuration(hiveConf);

    conf1.set("key1", "val");
    key1 = CachedClientPool.extractKey("conf:key1", conf1);
    key2 = CachedClientPool.extractKey("conf:key1", conf2);
    assertThat(key2)
        .as("Config with different values are not supposed to be equivalent")
        .isNotEqualTo(key1);

    conf2.set("key1", "val");
    conf2.set("key2", "val");
    key2 = CachedClientPool.extractKey("conf:key2", conf2);
    assertThat(key2)
        .as("Config with different keys are not supposed to be equivalent")
        .isNotEqualTo(key1);

    key1 = CachedClientPool.extractKey("conf:key1,ugi", conf1);
    key2 = CachedClientPool.extractKey("ugi,conf:key1", conf2);
    assertThat(key2).as("Config with same key/value should be equivalent").isEqualTo(key1);

    conf1.set("key2", "val");
    key1 = CachedClientPool.extractKey("conf:key2 ,conf:key1", conf1);
    key2 = CachedClientPool.extractKey("conf:key2,conf:key1", conf2);
    assertThat(key2).as("Config with same key/value should be equivalent").isEqualTo(key1);

    assertThatThrownBy(
        () -> CachedClientPool.extractKey("ugi,ugi", hiveConf),
        "Duplicate key elements should result in an error")
        .isInstanceOf(ValidationException.class)
        .hasMessageContaining("UGI key element already specified");

    assertThatThrownBy(
        () -> CachedClientPool.extractKey("conf:k1,conf:k2,CONF:k1", hiveConf),
        "Duplicate conf key elements should result in an error")
        .isInstanceOf(ValidationException.class)
        .hasMessageContaining("Conf key element k1 already specified");
  }