throw exception in case of error in connection creation

qubole · Jul 13, 2020 · 7ed5f01 · 7ed5f01
1 parent 4e19edc
commit 7ed5f01
Show file tree

Hide file tree

Showing 4 changed files with 15 additions and 10 deletions.
diff --git a/rubix-spi/src/main/java/com/qubole/rubix/spi/fop/ObjectFactory.java b/rubix-spi/src/main/java/com/qubole/rubix/spi/fop/ObjectFactory.java
@@ -21,7 +21,8 @@
  */
 public interface ObjectFactory<T>
 {
-  T create(String host, int socketTimeout, int connectTimeout);
+  T create(String host, int socketTimeout, int connectTimeout)
+          throws Exception;
 
   void destroy(T t);
 

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
@@ -56,11 +56,14 @@ public ObjectPoolPartition(ObjectPool<T> pool, PoolConfig config,
     this.log = new CustomLogger(name, host);
     this.takeSemaphore = new Semaphore(config.getMaxSize(), true);
     for (int i = 0; i < config.getMinSize(); i++) {
-      T object = objectFactory.create(host, socketTimeout, connectTimeout);
-      if (object != null) {
+      try {
+        T object = objectFactory.create(host, socketTimeout, connectTimeout);
         objectQueue.add(new Poolable<>(object, pool, host));
         totalCount++;
       }
+      catch (Exception e) {
+        log.warn("Error in initializing pool, error: " + e);
+      }
     }
   }
 
@@ -113,10 +116,8 @@ private Poolable<T> tryGetObject(Poolable<T> poolable)
     {
       try {
         T object = objectFactory.create(host, socketTimeout, connectTimeout);
-        if (object != null) {
-          poolable = new Poolable<>(object, pool, host);
-          totalCount++;
-        }
+        poolable = new Poolable<>(object, pool, host);
+        totalCount++;
         log.debug(String.format("Increased pool size by %d, to new size: %d, current queue size: %d, delta: %d",
                 1, totalCount, objectQueue.size(), 1));
       }

diff --git a/rubix-spi/src/main/java/com/qubole/rubix/spi/fop/SocketChannelObjectFactory.java b/rubix-spi/src/main/java/com/qubole/rubix/spi/fop/SocketChannelObjectFactory.java
@@ -37,9 +37,10 @@ public SocketChannelObjectFactory(int port)
 
   @Override
   public SocketChannel create(String host, int socketTimeout, int connectTimeout)
+          throws IOException
   {
     SocketAddress sad = new InetSocketAddress(host, this.port);
-    SocketChannel socket = null;
+    SocketChannel socket;
     try {
       socket = SocketChannel.open();
       socket.socket().setSoTimeout(socketTimeout);
@@ -49,6 +50,7 @@ public SocketChannel create(String host, int socketTimeout, int connectTimeout)
     }
     catch (IOException e) {
       log.warn(LDS_POOL + " : Unable to open connection to host " + host, e);
+      throw e;
     }
     return socket;
   }

diff --git a/rubix-spi/src/main/java/com/qubole/rubix/spi/fop/SocketObjectFactory.java b/rubix-spi/src/main/java/com/qubole/rubix/spi/fop/SocketObjectFactory.java
@@ -36,16 +36,17 @@ public SocketObjectFactory(int port)
 
   @Override
   public TSocket create(String host, int socketTimeout, int connectTimeout)
+          throws TTransportException
   {
     log.debug(BKS_POOL + " : Opening connection to host: " + host);
-    TSocket socket = null;
+    TSocket socket;
     try {
       socket = new TSocket(host, port, socketTimeout, connectTimeout);
       socket.open();
     }
     catch (TTransportException e) {
-      socket = null;
       log.warn("Unable to open connection to host " + host, e);
+      throw e;
     }
     return socket;
   }