private void saslNegotiate(Channel ch, String serverPrincipal) {
    assert eventLoop.inEventLoop();
    NettyFutureUtils.safeWriteAndFlush(ch, connectionHeaderPreamble.retainedDuplicate());
    UserGroupInformation ticket = provider.getRealUser(remoteId.getTicket());
    if (ticket == null) {
      saslFailInit(ch, serverPrincipal, new FatalConnectionException("ticket/user is null"));
      return;
    }
    Promise<Boolean> saslPromise = ch.eventLoop().newPromise();
    final NettyHBaseSaslRpcClientHandler saslHandler;
    try {
      saslHandler = new NettyHBaseSaslRpcClientHandler(saslPromise, ticket, provider, token,
        ((InetSocketAddress) ch.remoteAddress()).getAddress(), serverPrincipal,
        rpcClient.fallbackAllowed, this.rpcClient.conf);
    } catch (IOException e) {
      saslFailInit(ch, serverPrincipal, e);
      return;
    }
    ch.pipeline().addBefore(BufferCallBeforeInitHandler.NAME, null, new SaslChallengeDecoder())
      .addBefore(BufferCallBeforeInitHandler.NAME, NettyHBaseSaslRpcClientHandler.HANDLER_NAME,
        saslHandler);
    NettyFutureUtils.addListener(saslPromise, new FutureListener<Boolean>() {

      @Override
      public void operationComplete(Future<Boolean> future) throws Exception {
        if (future.isSuccess()) {
          ChannelPipeline p = ch.pipeline();
          // check if negotiate with server for connection header is necessary
          if (saslHandler.isNeedProcessConnectionHeader()) {
            Promise<Boolean> connectionHeaderPromise = ch.eventLoop().newPromise();
            // create the handler to handle the connection header
            NettyHBaseRpcConnectionHeaderHandler chHandler =
              new NettyHBaseRpcConnectionHeaderHandler(connectionHeaderPromise, conf,
                connectionHeaderWithLength);

            // add ReadTimeoutHandler to deal with server doesn't response connection header
            // because of the different configuration in client side and server side
            final String readTimeoutHandlerName = "ReadTimeout";
            p.addBefore(BufferCallBeforeInitHandler.NAME, readTimeoutHandlerName,
              new ReadTimeoutHandler(rpcClient.readTO, TimeUnit.MILLISECONDS))
              .addBefore(BufferCallBeforeInitHandler.NAME, null, chHandler);
            NettyFutureUtils.addListener(connectionHeaderPromise, new FutureListener<Boolean>() {
              @Override
              public void operationComplete(Future<Boolean> future) throws Exception {
                if (future.isSuccess()) {
                  ChannelPipeline p = ch.pipeline();
                  p.remove(readTimeoutHandlerName);
                  p.remove(NettyHBaseRpcConnectionHeaderHandler.class);
                  // don't send connection header, NettyHBaseRpcConnectionHeaderHandler
                  // sent it already
                  saslEstablished(ch, serverPrincipal);
                } else {
                  final Throwable error = future.cause();
                  scheduleRelogin(error);
                  saslFailInit(ch, serverPrincipal, toIOE(error));
                }
              }
            });
          } else {
            // send the connection header to server
            ch.write(connectionHeaderWithLength.retainedDuplicate());
            saslEstablished(ch, serverPrincipal);
          }
        } else {
          final Throwable error = future.cause();
          scheduleRelogin(error);
          saslFailInit(ch, serverPrincipal, toIOE(error));
        }
      }
    });
  }
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