Added rubix config for connection pool scavanger

- removed synchronized block from increaseObjects - refactored borrowObject method
qubole · Oct 1, 2020 · e6ff797 · e6ff797
1 parent 9383d53
commit e6ff797
Show file tree

Hide file tree

Showing 7 changed files with 8 additions and 9 deletions.
diff --git a/rubix-spi/src/main/java/com/qubole/rubix/spi/BookKeeperFactory.java b/rubix-spi/src/main/java/com/qubole/rubix/spi/BookKeeperFactory.java
@@ -72,7 +72,7 @@ public RetryingPooledBookkeeperClient createBookKeeperClient(String host, Config
     }
     else {
       Poolable<TTransport> obj;
-      obj = pool.borrowObject(host, conf);
+      obj = pool.borrowObject(host);
       RetryingPooledBookkeeperClient retryingBookkeeperClient = new RetryingPooledBookkeeperClient(obj, host, conf);
       return retryingBookkeeperClient;
     }

diff --git a/rubix-spi/src/main/java/com/qubole/rubix/spi/CacheConfig.java b/rubix-spi/src/main/java/com/qubole/rubix/spi/CacheConfig.java
@@ -52,6 +52,7 @@ public class CacheConfig
   private static final String KEY_POOL_MIN_SIZE = "rubix.pool.size.min";
   private static final String KEY_POOL_DELTA_SIZE = "rubix.pool.delta.size";
   private static final String KEY_POOL_MAX_WAIT_TIMEOUT = "rubix.pool.wait.timeout";
+  private static final String KEY_POOL_SCAVENGER_INTERVAL = "rubix.pool.scavenger.interval";
   private static final String KEY_DATA_CACHE_EXPIRY_AFTER_WRITE = "rubix.cache.expiration.after-write";
   private static final String KEY_DATA_CACHE_DIR_PREFIX = "rubix.cache.dirprefix.list";
   private static final String KEY_DATA_CACHE_DIR_SUFFIX = "rubix.cache.dirsuffix";
@@ -310,7 +311,7 @@ public static int getTransportPoolMaxWait(Configuration conf)
 
   public static int getScavengeInterval(Configuration conf)
   {
-    return conf.getInt(KEY_POOL_MAX_WAIT_TIMEOUT, DEFAULT_SCAVENGE_INTERVAL);
+    return conf.getInt(KEY_POOL_SCAVENGER_INTERVAL, DEFAULT_SCAVENGE_INTERVAL);
   }
 
   public static int get(Configuration conf)

diff --git a/rubix-spi/src/main/java/com/qubole/rubix/spi/DataTransferClientFactory.java b/rubix-spi/src/main/java/com/qubole/rubix/spi/DataTransferClientFactory.java
@@ -53,7 +53,7 @@ public static DataTransferClient getClient(String host, Configuration conf)
         }
       }
     }
-    Poolable<SocketChannel> socketChannelPoolable = pool.borrowObject(host, conf);
+    Poolable<SocketChannel> socketChannelPoolable = pool.borrowObject(host);
     return new DataTransferClient(socketChannelPoolable);
   }
 

diff --git a/rubix-spi/src/main/java/com/qubole/rubix/spi/RetryingPooledThriftClient.java b/rubix-spi/src/main/java/com/qubole/rubix/spi/RetryingPooledThriftClient.java
@@ -76,7 +76,7 @@ protected <V> V retryConnection(Callable<V> callable)
 
         // unset transportPoolable so that close() doesnt return it again to pool if borrowObject hits an exception
         transportPoolable = null;
-        transportPoolable = objectPool.borrowObject(host, conf);
+        transportPoolable = objectPool.borrowObject(host);
         updateClient(transportPoolable);
       }
     }

diff --git a/rubix-spi/src/main/java/com/qubole/rubix/spi/fop/ObjectPool.java b/rubix-spi/src/main/java/com/qubole/rubix/spi/fop/ObjectPool.java
@@ -67,7 +67,7 @@ protected BlockingQueue<Poolable<T>> createBlockingQueue(PoolConfig poolConfig)
     return new ArrayBlockingQueue<>(poolConfig.getMaxSize());
   }
 
-  public Poolable<T> borrowObject(String host, Configuration conf)
+  public Poolable<T> borrowObject(String host)
   {
     if (!hostToPoolMap.containsKey(host)) {
       synchronized (hostToPoolMap) {

diff --git a/rubix-spi/src/main/java/com/qubole/rubix/spi/fop/ObjectPoolPartition.java b/rubix-spi/src/main/java/com/qubole/rubix/spi/fop/ObjectPoolPartition.java
@@ -66,8 +66,6 @@ public void returnObject(Poolable<T> object)
     if (!objectFactory.validate(object.getObject())) {
       log.debug(String.format("Invalid object...removing: %s ", object));
       decreaseObject(object);
-      // Compensate for the removed object. Needed to prevent endless wait when in parallel a borrowObject is called
-      increaseObjects(1, false);
       return;
     }
 
@@ -115,7 +113,7 @@ public Poolable<T> getObject(boolean blocking)
     return freeObject;
   }
 
-  private synchronized Poolable<T> increaseObjects(int delta, boolean returnObject)
+  private Poolable<T> increaseObjects(int delta, boolean returnObject)
   {
     int oldCount = totalCount;
     if (delta + totalCount > config.getMaxSize()) {

diff --git a/rubix-spi/src/test/java/com/qubole/rubix/spi/client/TestPoolingClient.java b/rubix-spi/src/test/java/com/qubole/rubix/spi/client/TestPoolingClient.java
@@ -114,7 +114,7 @@ private RetryingPooledThriftTestClient getClient(int retries)
             retries,
             conf,
             "localhost",
-            pool.borrowObject("localhost", conf));
+            pool.borrowObject("localhost"));
   }
 
   private static void startServerAsync(final Configuration conf)