(gravitino) branch main updated: [#8483] improvement(core): prevent leaking unusable connections in ClientPoolImpl run method (#8547)

[yuqi4733]

This is an automated email from the ASF dual-hosted git repository.

yuqi4733 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git


The following commit(s) were added to refs/heads/main by this push:
     new f70a235227 [#8483] improvement(core): prevent leaking unusable 
connections in ClientPoolImpl run method (#8547)
f70a235227 is described below

commit f70a235227042f505cd6ec6982497b5e4c7961ca
Author: keepConcentration <phm...@gmail.com>
AuthorDate: Wed Sep 17 20:48:12 2025 +0900

    [#8483] improvement(core): prevent leaking unusable connections in 
ClientPoolImpl run method (#8547)
    
    ### What changes were proposed in this pull request?
    
    This PR updates the `ClientPoolImpl#run` method to ensure unusable
    connections are not leaked when a reconnection attempt fails.
    
    ### Why are the changes needed?
    
    Previously, if a reconnection attempt failed inside `run`, the old
    (unusable) client could still be released back into the pool, leading to
    potential connection leaks and invalid state.
    
    Fix: #8483
    
    ### Does this PR introduce _any_ user-facing change?
    
    No user-facing changes.
    
    ### How was this patch tested?
    
    Executed existing unit tests
---
 .../java/org/apache/gravitino/utils/ClientPoolImpl.java     | 13 +++++++++++--
 1 file changed, 11 insertions(+), 2 deletions(-)

diff --git a/core/src/main/java/org/apache/gravitino/utils/ClientPoolImpl.java 
b/core/src/main/java/org/apache/gravitino/utils/ClientPoolImpl.java
index 1ba2002946..5aa0528ee3 100644
--- a/core/src/main/java/org/apache/gravitino/utils/ClientPoolImpl.java
+++ b/core/src/main/java/org/apache/gravitino/utils/ClientPoolImpl.java
@@ -55,6 +55,7 @@ public abstract class ClientPoolImpl<C, E extends Exception>
   @Override
   public <R> R run(Action<R, C, E> action, boolean retry) throws E, 
InterruptedException {
     C client = get();
+    boolean shouldRelease = true;
     try {
       return action.run(client);
 
@@ -62,7 +63,13 @@ public abstract class ClientPoolImpl<C, E extends Exception>
       if (retry && isConnectionException(exc)) {
         try {
           client = reconnect(client);
-        } catch (Exception ignored) {
+        } catch (Exception reconnectException) {
+          shouldRelease = false;
+          synchronized (this) {
+            close(client);
+            currentSize -= 1;
+          }
+
           // if reconnection throws any exception, rethrow the original failure
           throw reconnectExc.cast(exc);
         }
@@ -73,7 +80,9 @@ public abstract class ClientPoolImpl<C, E extends Exception>
       throw exc;
 
     } finally {
-      release(client);
+      if (shouldRelease) {
+        release(client);
+      }
     }
   }