Method decode

src/RegionClient.java:1410–1585

Decodes the response of an RPC and triggers its Deferred. This method is only invoked by Netty from a Netty IO thread and will not be invoked on the same instance concurrently, so it only needs to be careful when accessing shared state, but certain synchronization constraints may be rela

(final ChannelHandlerContext ctx,
                          final Channel chan,
                          final ChannelBuffer channel_buffer,
                          final VoidEnum unused)

   * @return {@code null}, always.
   */
  @Override
  protected Object decode(final ChannelHandlerContext ctx,
                          final Channel chan,
                          final ChannelBuffer channel_buffer,
                          final VoidEnum unused) {
    ChannelBuffer buf = channel_buffer;
    final long start = System.nanoTime();
    final int rdx = buf.readerIndex();
    LOG.debug("------------------>> ENTERING DECODE >>------------------");
    final int rpcid;
    final RPCPB.ResponseHeader header;
    
    if (secure_rpc_helper != null) {
      buf = secure_rpc_helper.handleResponse(buf, chan);
      if (buf == null) {
        // everything in the buffer was part of the security handshake so we're
        // done here.
        return null;
      }
    }
    
    final int size;
    if (server_version >= SERVER_VERSION_095_OR_ABOVE) {
      size = buf.readInt();
      ensureReadable(buf, size);
      HBaseRpc.checkArrayLength(buf, size);
      header = HBaseRpc.readProtobuf(buf, RPCPB.ResponseHeader.PARSER);
      if (!header.hasCallId()) {
        final String msg = "RPC response (size: " + size + ") doesn't"
          + " have a call ID: " + header + ", buf=" + Bytes.pretty(buf);
        throw new NonRecoverableException(msg);
      }
      rpcid = header.getCallId();
    } else {  // HBase 0.94 and before.
      size = 0;
      header = null;  // No protobuf back then.
      rpcid = buf.readInt();
    }

    final HBaseRpc rpc = rpcs_inflight.get(rpcid);
    if (rpc == null) {
      // make sure to consume the RPC body so that we can decode the next
      // RPC in the buffer if there is one. Do this before incrementing counters
      // as with pre-protobuf RPCs we may throw replays while consuming.
      if (server_version >= SERVER_VERSION_095_OR_ABOVE) {
        buf.readerIndex(rdx + size + 4);
        if (LOG.isDebugEnabled()) {
          LOG.debug("Skipped timed out RPC ID " + rpcid + " of " + size + 
              " bytes on " + this);
        }
      } else {
        consumeTimedoutNonPBufRPC(buf, rdx, rpcid);
      }
      
      // TODO - account for the rpcid overflow
      if (rpcid > -1 && rpcid <= this.rpcid.get()) {
        if (LOG.isDebugEnabled()) {
          LOG.debug("Received a response for rpcid: " + rpcid +