Code changes in RPCserver for user provided port

Author: pralabhkumar

Diff for: rsc/src/main/java/com/cloudera/livy/rsc/rpc/RpcServer.java

@@ -19,7 +19,9 @@
 
 import java.io.Closeable;
 import java.io.IOException;
+import java.net.BindException;
 import java.net.InetSocketAddress;
+import java.net.ServerSocket;
 import java.security.SecureRandom;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
@@ -61,54 +63,91 @@ 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 int DEFAULT_RETRY=10;
+  
+  
   public RpcServer(RSCConf lconf) throws IOException, InterruptedException {
     this.config = lconf;
     this.group = new NioEventLoopGroup(
-        this.config.getInt(RPC_MAX_THREADS),
-        Utils.newDaemonThreadFactory("RPC-Handler-%d"));
-    this.channel = new ServerBootstrap()
-      .group(group)
-      .channel(NioServerSocketChannel.class)
-      .childHandler(new ChannelInitializer<SocketChannel>() {
-          @Override
-          public void initChannel(SocketChannel ch) throws Exception {
-            SaslServerHandler saslHandler = new SaslServerHandler(config);
-            final Rpc newRpc = Rpc.createServer(saslHandler, config, ch, group);
-            saslHandler.rpc = newRpc;
-
-            Runnable cancelTask = new Runnable() {
-                @Override
-                public void run() {
-                  LOG.warn("Timed out waiting for hello from client.");
-                  newRpc.close();
-                }
-            };
-            saslHandler.cancelTask = group.schedule(cancelTask,
-                config.getTimeAsMs(RPC_CLIENT_HANDSHAKE_TIMEOUT),
-                TimeUnit.MILLISECONDS);
-          }
-      })
-      .option(ChannelOption.SO_BACKLOG, 1)
-      .option(ChannelOption.SO_REUSEADDR, true)
-      .childOption(ChannelOption.SO_KEEPALIVE, true)
-      .bind(0)
-      .sync()
-      .channel();
+    this.config.getInt(RPC_MAX_THREADS),
+    Utils.newDaemonThreadFactory("RPC-Handler-%d"));
+    int portNumber=config.getInt(LAUNCHER_PORT);
+    for(int tries = 0 ; tries<DEFAULT_RETRY ; tries++){
+      try {
+        this.channel=createChannel(portNumber, tries);
+    		break;
+    	}catch(BindException e){
+    		LOG.warn("RPC not able to connect port "+ portNumber);
+    		portNumber = portNumber +1;
+    	}
+    }
+   
     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;
   }
+  
+  /**
+   * If user set the port number by livy.rsc.launcher.port then use that
+   * @param portNumber : Provided by the user 
+   * @return
+   * @throws IOException 
+   * @throws InterruptedException 
+   */
+  public Channel createChannel(int portNumber,int tries) throws IOException, InterruptedException{
+    if(portNumber==-1) {
+      return getChannel(0);
+	}
+	else {
+      return getChannel(config.getInt(LAUNCHER_PORT) + tries );
+	}
+  }
+  
+  /**
+   * @throws InterruptedException 
+   * 
+   */
+  public Channel getChannel(int portNumber) throws BindException, InterruptedException{
+	  Channel channel = new ServerBootstrap()
+		      .group(group)
+		      .channel(NioServerSocketChannel.class)
+		      .childHandler(new ChannelInitializer<SocketChannel>() {
+		          @Override
+		          public void initChannel(SocketChannel ch) throws Exception {
+		            SaslServerHandler saslHandler = new SaslServerHandler(config);
+		            final Rpc newRpc = Rpc.createServer(saslHandler, config, ch, group);
+		            saslHandler.rpc = newRpc;
+
+		            Runnable cancelTask = new Runnable() {
+		                @Override
+		                public void run() {
+		                  LOG.warn("Timed out waiting for hello from client.");
+		                  newRpc.close();
+		                }
+		            };
+		            saslHandler.cancelTask = group.schedule(cancelTask,
+		                config.getTimeAsMs(RPC_CLIENT_HANDSHAKE_TIMEOUT),
+		                TimeUnit.MILLISECONDS);
+		          }
+		      })
+		      .option(ChannelOption.SO_BACKLOG, 1)
+		      .option(ChannelOption.SO_REUSEADDR, true)
+		      .childOption(ChannelOption.SO_KEEPALIVE, true)
+		      .bind(portNumber)
+		      .sync()
+		      .channel(); 
+	  return channel;
+  }
 
   /**
    * Tells the RPC server to expect connections from clients.
@@ -310,3 +349,4 @@ private ClientInfo(String id, String secret, ClientCallback callback) {
   }
 
 }
+