[#9581] fix(hive): Perform resource cleanup in HiveClientPool close (#9617)

<!--
1. Title: [#<issue>] <type>(<scope>): <subject>
   Examples:
     - "[#123] feat(operator): support xxx"
     - "[#233] fix: check null before access result in xxx"
     - "[MINOR] refactor: fix typo in variable name"
     - "[MINOR] docs: fix typo in README"
     - "[#255] test: fix flaky test NameOfTheTest"
   Reference: https://www.conventionalcommits.org/en/v1.0.0/
2. If the PR is unfinished, please mark this PR as draft.
-->

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

This PR performs actual resource cleanup in `HiveClientPool.close`

### Why are the changes needed?

Any code path in the current implementation that close the pool via
`ClientPoolImpl.close()` will cause the connection leaks because the
HiveClientPool.close is a no-op.

Fix: #9581

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

No.

### How was this patch tested?

Add UTs.

Author: joeyutong

catalogs/hive-metastore-common/build.gradle.kts

@@ -120,6 +120,7 @@ dependencies {
     exclude("org.slf4j")
   }
   testImplementation(libs.junit.jupiter.api)
+  testImplementation(libs.mockito.core)
   testImplementation(libs.woodstox.core)
   testImplementation(libs.testcontainers)
   testImplementation(project(":integration-test-common", "testArtifacts"))

catalogs/hive-metastore-common/src/main/java/org/apache/gravitino/hive/HiveClientPool.java

@@ -41,7 +41,7 @@ public class HiveClientPool extends ClientPoolImpl<HiveClient, GravitinoRuntimeE
    * @param properties The configuration used to initialize the Hive Metastore clients.
    */
   public HiveClientPool(String name, int poolSize, Properties properties) {
-    // Do not allow retry by default as we rely on RetryingHiveClient
+    // Do not allow retry by default as we rely on RetryingMetaStoreClient
     super(poolSize, GravitinoRuntimeException.class, false);
     this.clientFactory = new HiveClientFactory(properties, name);
   }
@@ -58,17 +58,20 @@ protected HiveClient newClient() {
 
   @Override
   protected HiveClient reconnect(HiveClient client) {
+    // No-op reconnect: RetryingMetaStoreClient handles reconnect logic.
     LOG.warn("Reconnecting to Hive Metastore");
     return client;
   }
 
   @Override
   protected boolean isConnectionException(Exception e) {
+    // Pool-level reconnection is not required by design.
     return false;
   }
 
   @Override
   protected void close(HiveClient client) {
     LOG.info("Closing Hive Metastore client");
+    client.close();
   }
 }

catalogs/hive-metastore-common/src/test/java/org/apache/gravitino/hive/TestHiveClientPool.java

@@ -0,0 +1,100 @@
+/*
+ * 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.hive;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import com.google.common.collect.Lists;
+import java.util.List;
+import java.util.Properties;
+import org.apache.gravitino.exceptions.GravitinoRuntimeException;
+import org.apache.gravitino.hive.client.HiveClient;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+/**
+ * Referenced from Apache Iceberg's {@code TestHiveClientPool} implementation.
+ *
+ * <p>Source: hive-metastore/src/test/java/org/apache/iceberg/hive/TestHiveClientPool.java
+ */
+public class TestHiveClientPool {
+
+  private HiveClientPool clients;
+
+  @BeforeEach
+  public void before() {
+    HiveClientPool clientPool = new HiveClientPool("hive", 2, new Properties());
+    clients = Mockito.spy(clientPool);
+  }
+
+  @AfterEach
+  public void after() {
+    clients.close();
+    clients = null;
+  }
+
+  @Test
+  public void testNewClientFailure() {
+    Mockito.doThrow(new RuntimeException("Connection exception")).when(clients).newClient();
+    RuntimeException ex = assertThrows(RuntimeException.class, () -> clients.run(Object::toString));
+    assertEquals("Connection exception", ex.getMessage());
+  }
+
+  @Test
+  public void testReconnect() {
+    HiveClient hiveClient = newClient();
+
+    String metaMessage = "Got exception: org.apache.thrift.transport.TTransportException";
+    Mockito.doThrow(new GravitinoRuntimeException(metaMessage))
+        .when(hiveClient)
+        .getAllDatabases("");
+
+    GravitinoRuntimeException ex =
+        assertThrows(
+            GravitinoRuntimeException.class,
+            () -> clients.run(client -> client.getAllDatabases("")));
+    assertEquals("Got exception: org.apache.thrift.transport.TTransportException", ex.getMessage());
+    // Verify that the method is never called.
+    Mockito.verify(clients, Mockito.never()).reconnect(hiveClient);
+  }
+
+  @Test
+  public void testClose() throws Exception {
+    HiveClient hiveClient = newClient();
+
+    List<String> databases = Lists.newArrayList("db1", "db2");
+    Mockito.doReturn(databases).when(hiveClient).getAllDatabases("");
+    assertEquals(clients.run(client -> client.getAllDatabases("")), databases);
+
+    clients.close();
+    assertTrue(clients.isClosed());
+    Mockito.verify(hiveClient).close();
+  }
+
+  private HiveClient newClient() {
+    HiveClient hiveClient = Mockito.mock(HiveClient.class);
+    Mockito.doReturn(hiveClient).when(clients).newClient();
+    return hiveClient;
+  }
+}