[hive] Close metastore clients in hive catalog

paimon-common/src/main/java/org/apache/paimon/client/ClientPool.java

@@ -26,7 +26,7 @@
 import java.util.function.Supplier;
 
 /** Client pool for using multiple clients to execute actions. */
-public interface ClientPool<C, E extends Exception> {
+public interface ClientPool<C, E extends Exception> extends Closeable {
     /** Action interface with return object for client. */
     interface Action<R, C, E extends Exception> {
         R run(C client) throws E;
@@ -42,7 +42,7 @@ interface ExecuteAction<C, E extends Exception> {
     void execute(ExecuteAction<C, E> action) throws E, InterruptedException;
 
     /** Default implementation for {@link ClientPool}. */
-    abstract class ClientPoolImpl<C, E extends Exception> implements Closeable, ClientPool<C, E> {
+    abstract class ClientPoolImpl<C, E extends Exception> implements ClientPool<C, E> {
 
         private volatile LinkedBlockingDeque<C> clients;
 

paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java

@@ -1017,7 +1017,7 @@ public void repairTable(Identifier identifier) throws TableNotExistException {
 
     @Override
     public void close() throws Exception {
-        // do nothing
+        clients.close();
     }
 
     @Override

paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/pool/CachedClientPool.java

@@ -204,6 +204,14 @@ static Key extractKey(String clientClassName, String cacheKeys, Configuration co
         return Key.of(elements);
     }
 
+    @Override
+    public void close() throws IOException {
+        if (clientPoolCache != null) {
+            clientPoolCache.asMap().values().forEach(HiveClientPool::close);
+            clientPoolCache.cleanUp();
+        }
+    }
+
     static class Key {
         private final List<Object> elements;