public static KeyValue createKeyValueFromInputStream(InputStream in, boolean withTags)
    throws IOException {
    byte[] intBytes = new byte[Bytes.SIZEOF_INT];
    int bytesRead = 0;
    while (bytesRead < intBytes.length) {
      int n = in.read(intBytes, bytesRead, intBytes.length - bytesRead);
      if (n < 0) {
        if (bytesRead == 0) {
          throw new EOFException();
        }
        throw new IOException("Failed read of int, read " + bytesRead + " bytes");
      }
      bytesRead += n;
    }
    byte[] bytes = new byte[Bytes.toInt(intBytes)];
    IOUtils.readFully(in, bytes, 0, bytes.length);
    return withTags
      ? new KeyValue(bytes, 0, bytes.length)
      : new NoTagsKeyValue(bytes, 0, bytes.length);
  }
Create a KeyValue reading from the raw InputStream. Named
<code>createKeyValueFromInputStream</code> so doesn't clash with {@link #create(DataInput)}
@param in       inputStream to read.
@param withTags whether the keyvalue should include tags are not
@return Created KeyValue OR if we find a length of zero, we will return null which can be
        useful marking a stream as done.
hbase-common/src/main/java/org/apache/hadoop/hbase/KeyValueUtil.java