[apache#2525] fix(catalog-hive): Fix classNotFoundError in the Caffei…

…ne removal listener (apache#2622)

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

Close `HiveClientPool` before closing the class loader as the removal
listener is asynchronous and can't make this guarantee.

### Why are the changes needed?

It's bugs that need to be fixed.

Fix: apache#2525 

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

N/A.

### How was this patch tested?

Test locally.

Co-authored-by: Qi Yu <yuqi@datastrato.com>

catalogs/catalog-hive/src/main/java/com/datastrato/gravitino/catalog/hive/CachedClientPool.java

@@ -135,6 +135,13 @@ private static ThreadFactory newDaemonThreadFactory() {
   }
 
   public void close() {
+    // Close all the HiveClientPool instances in the cache first and then shutdown the scheduler. As
+    // Removal listener in Cache will be invoked by the scheduler asynchronously, we need to close
+    // the HiveClientPool instances first and then shutdown the scheduler. Another reason is that
+    // Caller may call this `close` method and then close the class loader that is needed by the
+    // `close` method. We must ensure that all the HiveClientPool instances are closed before the
+    // class loader is closed.
+    clientPoolCache.asMap().forEach((key, value) -> value.close());
     clientPoolCache.invalidateAll();
   }
 }

core/src/main/java/com/datastrato/gravitino/utils/ClientPoolImpl.java

@@ -36,7 +36,7 @@ public abstract class ClientPoolImpl<C, E extends Exception>
   private final Object signal = new Object();
   private final boolean retryByDefault;
   private volatile int currentSize;
-  private boolean closed;
+  private volatile boolean closed;
 
   public ClientPoolImpl(int poolSize, Class<? extends E> reconnectExc, boolean retryByDefault) {
     this.poolSize = poolSize;
@@ -89,6 +89,12 @@ protected boolean isConnectionException(Exception exc) {
 
   @Override
   public void close() {
+    // To avoid closing it repeatedly, we add a judgment that if it has been closed,
+    // we do not need to close it anymore.
+    if (closed) {
+      return;
+    }
+
     this.closed = true;
     try {
       while (currentSize > 0) {