private void connect(Call connectionRegistryCall) throws UnknownHostException {
    assert eventLoop.inEventLoop();
    LOG.trace("Connecting to {}", remoteId.getAddress());
    InetSocketAddress remoteAddr = getRemoteInetAddress(rpcClient.metrics);
    this.channel = new Bootstrap().group(eventLoop).channel(rpcClient.channelClass)
      .option(ChannelOption.TCP_NODELAY, rpcClient.isTcpNoDelay())
      .option(ChannelOption.SO_KEEPALIVE, rpcClient.tcpKeepAlive)
      .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, rpcClient.connectTO)
      .handler(new ChannelInitializer<Channel>() {
        @Override
        protected void initChannel(Channel ch) throws Exception {
          if (conf.getBoolean(X509Util.HBASE_CLIENT_NETTY_TLS_ENABLED, false)) {
            SslContext sslContext = rpcClient.getSslContext();
            SslHandler sslHandler = sslContext.newHandler(ch.alloc(),
              remoteId.address.getHostName(), remoteId.address.getPort());
            sslHandler.setHandshakeTimeoutMillis(
              conf.getInt(X509Util.HBASE_CLIENT_NETTY_TLS_HANDSHAKETIMEOUT,
                X509Util.DEFAULT_HANDSHAKE_DETECTION_TIMEOUT_MILLIS));
            ch.pipeline().addFirst(sslHandler);
            LOG.debug("SSL handler added with handshake timeout {} ms",
              sslHandler.getHandshakeTimeoutMillis());
          }
          ch.pipeline().addLast(BufferCallBeforeInitHandler.NAME,
            new BufferCallBeforeInitHandler());
        }
      }).localAddress(rpcClient.localAddr).remoteAddress(remoteAddr).connect()
      .addListener(new ChannelFutureListener() {

        private void succeed(Channel ch) throws IOException {
          if (connectionRegistryCall != null) {
            getConnectionRegistry(ch, connectionRegistryCall);
            return;
          }
          if (!useSasl) {
            // BufferCallBeforeInitHandler will call ctx.flush when receiving the
            // BufferCallEvent.success() event, so here we just use write for the below two messages
            NettyFutureUtils.safeWrite(ch, connectionHeaderPreamble.retainedDuplicate());
            NettyFutureUtils.safeWrite(ch, connectionHeaderWithLength.retainedDuplicate());
            established(ch);
          } else {
            saslNegotiate(ch);
          }
        }

        private void fail(Channel ch, Throwable error) {
          IOException ex = toIOE(error);
          LOG.warn("Exception encountered while connecting to the server " + remoteId.getAddress(),
            ex);
          if (connectionRegistryCall != null) {
            connectionRegistryCall.setException(ex);
          }
          failInit(ch, ex);
        }

        @Override
        public void operationComplete(ChannelFuture future) throws Exception {
          Channel ch = future.channel();
          if (!future.isSuccess()) {
            fail(ch, future.cause());
            return;
          }
          SslHandler sslHandler = ch.pipeline().get(SslHandler.class);
          if (sslHandler != null) {
            NettyFutureUtils.addListener(sslHandler.handshakeFuture(), f -> {
              if (f.isSuccess()) {
                succeed(ch);
              } else {
                fail(ch, f.cause());
              }
            });
          } else {
            succeed(ch);
          }
        }
      }).channel();
  }
RPC connection implementation based on netty.
<p/>
Most operations are executed in handlers. Netty handler is always executed in the same
thread(EventLoop) so no lock is needed.
<p/>
<strong>Implementation assumptions:</strong> All the private methods should be called in the
{@link #eventLoop} thread, otherwise there will be races.
@since 2.0.0
hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/NettyRpcConnection.java