public int readAndProcess() throws IOException, InterruptedException {
    // If we have not read the connection setup preamble, look to see if that is on the wire.
    if (!connectionPreambleRead) {
      int count = readPreamble();
      if (!connectionPreambleRead) {
        return count;
      }
    }

    // Try and read in an int. it will be length of the data to read (or -1 if a ping). We catch the
    // integer length into the 4-byte this.dataLengthBuffer.
    int count = read4Bytes();
    if (count < 0 || dataLengthBuffer.remaining() > 0) {
      return count;
    }

    // We have read a length and we have read the preamble. It is either the connection header
    // or it is a request.
    if (data == null) {
      dataLengthBuffer.flip();
      int dataLength = dataLengthBuffer.getInt();
      if (dataLength == RpcClient.PING_CALL_ID) {
        if (!useWrap) { // covers the !useSasl too
          dataLengthBuffer.clear();
          return 0; // ping message
        }
      }
      if (dataLength < 0) { // A data length of zero is legal.
        throw new DoNotRetryIOException(
          "Unexpected data length " + dataLength + "!! from " + getHostAddress());
      }

      if (dataLength > this.rpcServer.maxRequestSize) {
        String msg = "RPC data length of " + dataLength + " received from " + getHostAddress()
          + " is greater than max allowed " + this.rpcServer.maxRequestSize + ". Set \""
          + SimpleRpcServer.MAX_REQUEST_SIZE
          + "\" on server to override this limit (not recommended)";
        SimpleRpcServer.LOG.warn(msg);

        if (connectionHeaderRead && connectionPreambleRead) {
          incRpcCount();
          // Construct InputStream for the non-blocking SocketChannel
          // We need the InputStream because we want to read only the request header
          // instead of the whole rpc.
          ByteBuffer buf = ByteBuffer.allocate(1);
          InputStream is = new InputStream() {
            @Override
            public int read() throws IOException {
              SimpleServerRpcConnection.this.rpcServer.channelRead(channel, buf);
              buf.flip();
              int x = buf.get();
              buf.flip();
              return x;
            }
          };
          CodedInputStream cis = CodedInputStream.newInstance(is);
          int headerSize = cis.readRawVarint32();
          Message.Builder builder = RequestHeader.newBuilder();
          ProtobufUtil.mergeFrom(builder, cis, headerSize);
          RequestHeader header = (RequestHeader) builder.build();

          // Notify the client about the offending request
          SimpleServerCall reqTooBig = new SimpleServerCall(header.getCallId(), this.service, null,
            null, null, null, this, 0, this.addr, EnvironmentEdgeManager.currentTime(), 0,
            this.rpcServer.bbAllocator, this.rpcServer.cellBlockBuilder, null, responder);
          RequestTooBigException reqTooBigEx = new RequestTooBigException(msg);
          this.rpcServer.metrics.exception(reqTooBigEx);
          // Make sure the client recognizes the underlying exception
          // Otherwise, throw a DoNotRetryIOException.
          if (
            VersionInfoUtil.hasMinimumVersion(connectionHeader.getVersionInfo(),
              RequestTooBigException.MAJOR_VERSION, RequestTooBigException.MINOR_VERSION)
          ) {
            reqTooBig.setResponse(null, null, reqTooBigEx, msg);
          } else {
            reqTooBig.setResponse(null, null, new DoNotRetryIOException(msg), msg);
          }
          // In most cases we will write out the response directly. If not, it is still OK to just
          // close the connection without writing out the reqTooBig response. Do not try to write
          // out directly here, and it will cause deserialization error if the connection is slow
          // and we have a half writing response in the queue.
          reqTooBig.sendResponseIfReady();
        }
        // Close the connection
        return -1;
      }

      // Initialize this.data with a ByteBuff.
      // This call will allocate a ByteBuff to read request into and assign to this.data
      // Also when we use some buffer(s) from pool, it will create a CallCleanup instance also and
      // assign to this.callCleanup
      initByteBuffToReadInto(dataLength);

      // Increment the rpc count. This counter will be decreased when we write
      // the response. If we want the connection to be detected as idle properly, we
      // need to keep the inc / dec correct.
      incRpcCount();
    }

    count = channelDataRead(channel, data);

    if (count >= 0 && data.remaining() == 0) { // count==0 if dataLength == 0
      process();
    }

    return count;
  }
Read off the wire. If there is not enough data to read, update the connection state with what
we have and returns.
@return Returns -1 if failure (and caller will close connection), else zero or more.
hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/SimpleServerRpcConnection.java