private void writeDataAndReadFromHFile(Path hfilePath, Compression.Algorithm compressAlgo,
    int entryCount, boolean findMidKey, boolean useTags) throws IOException {

    HFileContext context = new HFileContextBuilder().withBlockSize(4096).withIncludesTags(useTags)
      .withDataBlockEncoding(dataBlockEncoding).withCellComparator(CellComparatorImpl.COMPARATOR)
      .withCompression(compressAlgo).build();
    CacheConfig cacheConfig = new CacheConfig(conf);
    HFile.Writer writer = new HFile.WriterFactory(conf, cacheConfig).withPath(fs, hfilePath)
      .withFileContext(context).create();

    List<KeyValue> keyValues = new ArrayList<>(entryCount);
    writeKeyValues(entryCount, useTags, writer, RNG, keyValues);

    FSDataInputStream fsdis = fs.open(hfilePath);

    long fileSize = fs.getFileStatus(hfilePath).getLen();
    FixedFileTrailer trailer = FixedFileTrailer.readFromStream(fsdis, fileSize);

    Assert.assertEquals(3, trailer.getMajorVersion());
    Assert.assertEquals(entryCount, trailer.getEntryCount());
    HFileContext meta = new HFileContextBuilder().withCompression(compressAlgo)
      .withIncludesMvcc(true).withIncludesTags(useTags).withDataBlockEncoding(dataBlockEncoding)
      .withHBaseCheckSum(true).build();
    ReaderContext readerContext =
      new ReaderContextBuilder().withInputStreamWrapper(new FSDataInputStreamWrapper(fsdis))
        .withFilePath(hfilePath).withFileSystem(fs).withFileSize(fileSize).build();
    HFileBlock.FSReader blockReader =
      new HFileBlock.FSReaderImpl(readerContext, meta, ByteBuffAllocator.HEAP, conf);
    // Comparator class name is stored in the trailer in version 3.
    CellComparator comparator = trailer.createComparator();
    HFileBlockIndex.BlockIndexReader dataBlockIndexReader =
      new HFileBlockIndex.CellBasedKeyBlockIndexReaderV2(comparator,
        trailer.getNumDataIndexLevels());
    HFileBlockIndex.BlockIndexReader metaBlockIndexReader =
      new HFileBlockIndex.ByteArrayKeyBlockIndexReader(1);

    HFileBlock.BlockIterator blockIter = blockReader.blockRange(trailer.getLoadOnOpenDataOffset(),
      fileSize - trailer.getTrailerSize());
    // Data index. We also read statistics about the block index written after
    // the root level.
    dataBlockIndexReader.readMultiLevelIndexRoot(
      blockIter.nextBlockWithBlockType(BlockType.ROOT_INDEX), trailer.getDataIndexCount());

    FSDataInputStreamWrapper wrapper = new FSDataInputStreamWrapper(fs, hfilePath);
    readerContext = new ReaderContextBuilder().withFilePath(hfilePath).withFileSize(fileSize)
      .withFileSystem(wrapper.getHfs()).withInputStreamWrapper(wrapper).build();
    HFileInfo hfile = new HFileInfo(readerContext, conf);
    HFile.Reader reader = new HFilePreadReader(readerContext, hfile, cacheConfig, conf);
    hfile.initMetaAndIndex(reader);
    if (findMidKey) {
      Cell midkey = dataBlockIndexReader.midkey(reader);
      Assert.assertNotNull("Midkey should not be null", midkey);
    }

    // Meta index.
    metaBlockIndexReader.readRootIndex(
      blockIter.nextBlockWithBlockType(BlockType.ROOT_INDEX).getByteStream(),
      trailer.getMetaIndexCount());
    // File info
    HFileInfo fileInfo = new HFileInfo();
    fileInfo.read(blockIter.nextBlockWithBlockType(BlockType.FILE_INFO).getByteStream());
    byte[] keyValueFormatVersion = fileInfo.get(HFileWriterImpl.KEY_VALUE_VERSION);
    boolean includeMemstoreTS =
      keyValueFormatVersion != null && Bytes.toInt(keyValueFormatVersion) > 0;

    // Counters for the number of key/value pairs and the number of blocks
    int entriesRead = 0;
    int blocksRead = 0;
    long memstoreTS = 0;

    DataBlockEncoder encoder = dataBlockEncoding.getEncoder();
    long curBlockPos = scanBlocks(entryCount, context, keyValues, fsdis, trailer, meta, blockReader,
      entriesRead, blocksRead, encoder);

    // Meta blocks. We can scan until the load-on-open data offset (which is
    // the root block index offset in version 2) because we are not testing
    // intermediate-level index blocks here.

    int metaCounter = 0;
    while (fsdis.getPos() < trailer.getLoadOnOpenDataOffset()) {
      LOG.info("Current offset: {}, scanning until {}", fsdis.getPos(),
        trailer.getLoadOnOpenDataOffset());
      HFileBlock block =
        blockReader.readBlockData(curBlockPos, -1, false, false, true).unpack(context, blockReader);
      Assert.assertEquals(BlockType.META, block.getBlockType());
      Text t = new Text();
      ByteBuff buf = block.getBufferWithoutHeader();
      if (Writables.getWritable(buf.array(), buf.arrayOffset(), buf.limit(), t) == null) {
        throw new IOException(
          "Failed to deserialize block " + this + " into a " + t.getClass().getSimpleName());
      }
      Text expectedText = (metaCounter == 0 ? new Text("Paris")
        : metaCounter == 1 ? new Text("Moscow")
        : new Text("Washington, D.C."));
      Assert.assertEquals(expectedText, t);
      LOG.info("Read meta block data: " + t);
      ++metaCounter;
      curBlockPos += block.getOnDiskSizeWithHeader();
    }

    fsdis.close();
    reader.close();
  }
Testing writing a version 3 {@link HFile} for all encoded blocks
hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestHFileWriterV3WithDataEncoders.java