@Override
    protected ExtendedCell parseCell() throws IOException {
      int keylength = StreamUtils.readRawVarint32(in);
      int vlength = StreamUtils.readRawVarint32(in);
      int tagsLength = StreamUtils.readRawVarint32(in);
      int length = 0;
      if (tagsLength == 0) {
        length = KeyValue.KEYVALUE_INFRASTRUCTURE_SIZE + keylength + vlength;
      } else {
        length = KeyValue.KEYVALUE_WITH_TAGS_INFRASTRUCTURE_SIZE + keylength + vlength + tagsLength;
      }

      byte[] backingArray = new byte[length];
      int pos = 0;
      pos = Bytes.putInt(backingArray, pos, keylength);
      pos = Bytes.putInt(backingArray, pos, vlength);

      // the row
      int elemLen = readIntoArray(backingArray, pos + Bytes.SIZEOF_SHORT,
        compression.getDictionary(CompressionContext.DictionaryIndex.ROW));
      checkLength(elemLen, Short.MAX_VALUE);
      pos = Bytes.putShort(backingArray, pos, (short) elemLen);
      pos += elemLen;

      // family
      elemLen = readIntoArray(backingArray, pos + Bytes.SIZEOF_BYTE,
        compression.getDictionary(CompressionContext.DictionaryIndex.FAMILY));
      checkLength(elemLen, Byte.MAX_VALUE);
      pos = Bytes.putByte(backingArray, pos, (byte) elemLen);
      pos += elemLen;

      // qualifier
      elemLen = readIntoArray(backingArray, pos,
        compression.getDictionary(CompressionContext.DictionaryIndex.QUALIFIER));
      pos += elemLen;

      // timestamp
      long ts = StreamUtils.readLong(in);
      pos = Bytes.putLong(backingArray, pos, ts);
      // type and value
      int typeValLen = length - pos;
      if (tagsLength > 0) {
        typeValLen = typeValLen - tagsLength - KeyValue.TAGS_LENGTH_SIZE;
      }
      pos = Bytes.putByte(backingArray, pos, (byte) in.read());
      int valLen = typeValLen - 1;
      if (hasValueCompression) {
        readCompressedValue(in, backingArray, pos, valLen);
        pos += valLen;
      } else {
        IOUtils.readFully(in, backingArray, pos, valLen);
        pos += valLen;
      }
      // tags
      if (tagsLength > 0) {
        pos = Bytes.putAsShort(backingArray, pos, tagsLength);
        if (hasTagCompression) {
          compression.tagCompressionContext.uncompressTags(in, backingArray, pos, tagsLength);
        } else {
          IOUtils.readFully(in, backingArray, pos, tagsLength);
        }
      }
      return new KeyValue(backingArray, 0, length);
    }
Create and setup a {@link WALCellCodec} from the CompressionContext. Cell Codec classname is
read from {@link Configuration}. Fully prepares the codec for use.
@param conf        {@link Configuration} to read for the user-specified codec. If none is
                   specified, uses a {@link WALCellCodec}.
@param compression compression the codec should use
@return a {@link WALCellCodec} ready for use.
@throws UnsupportedOperationException if the codec cannot be instantiated
hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/WALCellCodec.java