[#5442] Improvement(iceberg-common): Overwrite the equals and hashCod…

…e methods to avoid frequently creating HiveClientPool instances (#5478)

### What changes were proposed in this pull request?

Overwrite the equals and hashCode methods to avoid frequently creating
HiveClientPool instances

### Why are the changes needed?

Fix: #5442

### Does this PR introduce _any_ user-facing change?

No.

### How was this patch tested?

New UT.

Co-authored-by: cai can <[email protected]>
Co-authored-by: caican <[email protected]>

iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/utils/IcebergHiveCachedClientPool.java

@@ -33,6 +33,7 @@
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
@@ -47,6 +48,8 @@
 import org.apache.iceberg.util.PropertyUtil;
 import org.apache.iceberg.util.ThreadPools;
 import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * Referred from Apache Iceberg's CachedClientPool implementation
@@ -78,6 +81,8 @@
  */
 public class IcebergHiveCachedClientPool
     implements ClientPool<IMetaStoreClient, TException>, Closeable {
+  private static final Logger LOG = LoggerFactory.getLogger(IcebergHiveCachedClientPool.class);
+
   private static final String CONF_ELEMENT_PREFIX = "conf:";
 
   private static Cache<Key, HiveClientPool> clientPoolCache;
@@ -107,7 +112,13 @@ public IcebergHiveCachedClientPool(Configuration conf, Map<String, String> prope
   @VisibleForTesting
   HiveClientPool clientPool() {
     Key key = extractKey(properties.get(CatalogProperties.CLIENT_POOL_CACHE_KEYS), conf);
-    return clientPoolCache.get(key, k -> new HiveClientPool(clientPoolSize, conf));
+    return clientPoolCache.get(
+        key,
+        k -> {
+          HiveClientPool hiveClientPool = new HiveClientPool(clientPoolSize, conf);
+          LOG.info("Created a new HiveClientPool instance: {} for Key: {}", hiveClientPool, key);
+          return hiveClientPool;
+        });
   }
 
   private synchronized void init() {
@@ -118,7 +129,17 @@ private synchronized void init() {
       clientPoolCache =
           Caffeine.newBuilder()
               .expireAfterAccess(evictionInterval, TimeUnit.MILLISECONDS)
-              .removalListener((ignored, value, cause) -> ((HiveClientPool) value).close())
+              .removalListener(
+                  (key, value, cause) -> {
+                    HiveClientPool hiveClientPool = (HiveClientPool) value;
+                    if (hiveClientPool != null) {
+                      LOG.info(
+                          "Removing an expired HiveClientPool instance: {} for Key: {}",
+                          hiveClientPool,
+                          key);
+                      hiveClientPool.close();
+                    }
+                  })
               .scheduler(Scheduler.forScheduledExecutorService(scheduledExecutorService))
               .build();
     }
@@ -211,6 +232,23 @@ public Key(List<Object> elements) {
     static Key of(List<Object> elements) {
       return new Key(elements);
     }
+
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) {
+        return true;
+      }
+      if (o == null || !(o instanceof Key)) {
+        return false;
+      }
+      Key key = (Key) o;
+      return Objects.equals(elements, key.elements);
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hash(elements);
+    }
   }
 
   private enum KeyElementType {

iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/utils/TestIcebergHiveCachedClientPool.java

@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.gravitino.iceberg.common.utils;
+
+import com.google.common.collect.Maps;
+import java.io.IOException;
+import java.security.PrivilegedAction;
+import java.util.Map;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.iceberg.hive.HiveClientPool;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class TestIcebergHiveCachedClientPool {
+
+  @Test
+  void test() throws IOException {
+    Configuration configuration = new Configuration();
+    configuration.set("hive.metastore.uris", "thrift://localhost:9083");
+    Map<String, String> properties = Maps.newHashMap();
+    IcebergHiveCachedClientPool clientPool =
+        new IcebergHiveCachedClientPool(configuration, properties);
+
+    // test extractKey for simple conf
+    IcebergHiveCachedClientPool.Key key1 =
+        IcebergHiveCachedClientPool.extractKey(null, configuration);
+    IcebergHiveCachedClientPool.Key key2 =
+        IcebergHiveCachedClientPool.extractKey(null, configuration);
+    Assertions.assertEquals(key1, key2);
+
+    // test clientPool
+    HiveClientPool hiveClientPool1 = clientPool.clientPool();
+    HiveClientPool hiveClientPool2 = clientPool.clientPool();
+    Assertions.assertEquals(hiveClientPool1, hiveClientPool2);
+
+    // test extractKey with user_name or ugi
+    UserGroupInformation current = UserGroupInformation.getCurrentUser();
+    UserGroupInformation foo1 = UserGroupInformation.createProxyUser("foo", current);
+    UserGroupInformation foo2 = UserGroupInformation.createProxyUser("foo", current);
+    UserGroupInformation bar = UserGroupInformation.createProxyUser("bar", current);
+
+    IcebergHiveCachedClientPool.Key key3 =
+        foo1.doAs(
+            (PrivilegedAction<IcebergHiveCachedClientPool.Key>)
+                () -> IcebergHiveCachedClientPool.extractKey("user_name", configuration));
+    IcebergHiveCachedClientPool.Key key4 =
+        foo2.doAs(
+            (PrivilegedAction<IcebergHiveCachedClientPool.Key>)
+                () -> IcebergHiveCachedClientPool.extractKey("user_name", configuration));
+    Assertions.assertEquals(key3, key4);
+
+    IcebergHiveCachedClientPool.Key key5 =
+        foo1.doAs(
+            (PrivilegedAction<IcebergHiveCachedClientPool.Key>)
+                () -> IcebergHiveCachedClientPool.extractKey("user_name", configuration));
+    IcebergHiveCachedClientPool.Key key6 =
+        bar.doAs(
+            (PrivilegedAction<IcebergHiveCachedClientPool.Key>)
+                () -> IcebergHiveCachedClientPool.extractKey("user_name", configuration));
+    Assertions.assertNotEquals(key5, key6);
+
+    IcebergHiveCachedClientPool.Key key7 =
+        foo1.doAs(
+            (PrivilegedAction<IcebergHiveCachedClientPool.Key>)
+                () -> IcebergHiveCachedClientPool.extractKey("ugi", configuration));
+    IcebergHiveCachedClientPool.Key key8 =
+        foo2.doAs(
+            (PrivilegedAction<IcebergHiveCachedClientPool.Key>)
+                () -> IcebergHiveCachedClientPool.extractKey("ugi", configuration));
+    Assertions.assertNotEquals(key7, key8);
+
+    // The equals method of UserGroupInformation: return this.subject ==
+    // ((UserGroupInformation)o).subject;
+    IcebergHiveCachedClientPool.Key key9 =
+        foo1.doAs(
+            (PrivilegedAction<IcebergHiveCachedClientPool.Key>)
+                () -> IcebergHiveCachedClientPool.extractKey("ugi", configuration));
+    IcebergHiveCachedClientPool.Key key10 =
+        bar.doAs(
+            (PrivilegedAction<IcebergHiveCachedClientPool.Key>)
+                () -> IcebergHiveCachedClientPool.extractKey("ugi", configuration));
+    Assertions.assertNotEquals(key9, key10);
+  }
+}