Livy:337 Binding RPCServer to user provided port and not random port

rsc/src/main/java/com/cloudera/livy/rsc/RSCConf.java

@@ -51,8 +51,10 @@ public static enum Entry implements ConfEntry {
 
     // Address for the RSC driver to connect back with it's connection info.
     LAUNCHER_ADDRESS("launcher.address", null),
+    LAUNCHER_PORT_RANGE("launcher.port.range", "10000~10110"),
+    // Setting up of this propety by user has no benefit. It is currently being used
+    // to pass  port information from ContextLauncher to RSCDriver
     LAUNCHER_PORT("launcher.port", -1),
-
     // How long will the RSC wait for a connection for a Livy server before shutting itself down.
     SERVER_IDLE_TIMEOUT("server.idle-timeout", "10m"),
 

rsc/src/main/java/com/cloudera/livy/rsc/rpc/RpcServer.java

@@ -19,7 +19,10 @@
 
 import java.io.Closeable;
 import java.io.IOException;
+import java.net.BindException;
 import java.net.InetSocketAddress;
+import java.net.ServerSocket;
+import java.net.SocketException;
 import java.security.SecureRandom;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
@@ -61,18 +64,78 @@ public class RpcServer implements Closeable {
   private static final SecureRandom RND = new SecureRandom();
 
   private final String address;
-  private final Channel channel;
+  private Channel channel;
   private final EventLoopGroup group;
   private final int port;
   private final ConcurrentMap<String, ClientInfo> pendingClients;
   private final RSCConf config;
-
+  private final String portRange;
+  private static enum PortRangeSchema{START_PORT, END_PORT, max};
+  private final String PORT_DELIMITER = "~";
+  /**
+   * Creating RPC Server
+   * @param lconf
+   * @throws IOException
+   * @throws InterruptedException
+   */
   public RpcServer(RSCConf lconf) throws IOException, InterruptedException {
     this.config = lconf;
+    this.portRange = config.get(LAUNCHER_PORT_RANGE);
     this.group = new NioEventLoopGroup(
-        this.config.getInt(RPC_MAX_THREADS),
-        Utils.newDaemonThreadFactory("RPC-Handler-%d"));
-    this.channel = new ServerBootstrap()
+    this.config.getInt(RPC_MAX_THREADS),
+    Utils.newDaemonThreadFactory("RPC-Handler-%d"));
+    int [] portData = getPortNumberAndRange();
+    int startingPortNumber = portData[PortRangeSchema.START_PORT.ordinal()];
+    int endPort = portData[PortRangeSchema.END_PORT.ordinal()];
+    boolean isContected = false;
+    for(int tries = startingPortNumber ; tries<=endPort ; tries++){
+      try {
+        this.channel = getChannel(tries);
+        isContected = true;
+        break;
+      }catch(SocketException e){
+        LOG.warn("RPC not able to connect port " + tries + " " + e.getMessage());
+      }
+    }
+    if(!isContected) {
+      throw new IOException("Unable to connect to provided ports " + this.portRange);
+    }
+    this.port = ((InetSocketAddress) channel.localAddress()).getPort();
+    this.pendingClients = new ConcurrentHashMap<>();
+    LOG.warn("Connected to the port " + this.port);
+    String address = config.get(RPC_SERVER_ADDRESS);
+    if (address == null) {
+      address = config.findLocalAddress();
+    }
+    this.address = address;
+  }
+
+  /**
+   * Get Port Numbers
+   */
+  public int[] getPortNumberAndRange() throws ArrayIndexOutOfBoundsException, NumberFormatException{
+    String[] split = this.portRange.split(PORT_DELIMITER);
+    int [] portRange=new int [PortRangeSchema.max.ordinal()];
+    try {
+      portRange[PortRangeSchema.START_PORT.ordinal()] =
+      Integer.parseInt(split[PortRangeSchema.START_PORT.ordinal()]);
+      portRange[PortRangeSchema.END_PORT.ordinal()] =
+      Integer.parseInt(split[PortRangeSchema.END_PORT.ordinal()]);
+    }catch(ArrayIndexOutOfBoundsException e) {
+      LOG.error("Port Range format is not correct " + this.portRange);
+      throw e;
+    }
+    catch(NumberFormatException e) {
+      LOG.error("Port are not in numeric format " + this.portRange);
+      throw e;
+    }
+    return portRange;
+  }
+  /**
+   * @throws InterruptedException
+   **/
+  public Channel getChannel(int portNumber) throws BindException, InterruptedException{
+      Channel channel = new ServerBootstrap()
       .group(group)
       .channel(NioServerSocketChannel.class)
       .childHandler(new ChannelInitializer<SocketChannel>() {
@@ -97,19 +160,11 @@ public void run() {
       .option(ChannelOption.SO_BACKLOG, 1)
       .option(ChannelOption.SO_REUSEADDR, true)
       .childOption(ChannelOption.SO_KEEPALIVE, true)
-      .bind(0)
+      .bind(portNumber)
       .sync()
       .channel();
-    this.port = ((InetSocketAddress) channel.localAddress()).getPort();
-    this.pendingClients = new ConcurrentHashMap<>();
-
-    String address = config.get(RPC_SERVER_ADDRESS);
-    if (address == null) {
-      address = config.findLocalAddress();
-    }
-    this.address = address;
+  return channel;
   }
-
   /**
    * Tells the RPC server to expect connections from clients.
    *
@@ -310,3 +365,4 @@ private ClientInfo(String id, String secret, ClientCallback callback) {
   }
 
 }
+