protected void testChecksumCorruptionInternals(boolean useTags) throws IOException {
    for (Compression.Algorithm algo : COMPRESSION_ALGORITHMS) {
      for (boolean pread : new boolean[] { false, true }) {
        LOG.info("testChecksumCorruption: Compression algorithm: " + algo + ", pread=" + pread);
        Path path = new Path(TEST_UTIL.getDataTestDir(), "blocks_v2_" + algo);
        FSDataOutputStream os = fs.create(path);
        HFileContext meta = new HFileContextBuilder().withCompression(algo).withIncludesMvcc(true)
          .withIncludesTags(useTags).withBytesPerCheckSum(HFile.DEFAULT_BYTES_PER_CHECKSUM).build();
        HFileBlock.Writer hbw = new HFileBlock.Writer(TEST_UTIL.getConfiguration(), null, meta);
        long totalSize = 0;
        for (int blockId = 0; blockId < 2; ++blockId) {
          DataOutputStream dos = hbw.startWriting(BlockType.DATA);
          for (int i = 0; i < 1234; ++i)
            dos.writeInt(i);
          hbw.writeHeaderAndData(os);
          totalSize += hbw.getOnDiskSizeWithHeader();
        }
        os.close();

        // Use hbase checksums.
        assertEquals(true, hfs.useHBaseChecksum());

        // Do a read that purposely introduces checksum verification failures.
        FSDataInputStreamWrapper is = new FSDataInputStreamWrapper(fs, path);
        meta = new HFileContextBuilder().withCompression(algo).withIncludesMvcc(true)
          .withIncludesTags(useTags).withHBaseCheckSum(true).build();
        ReaderContext context = new ReaderContextBuilder().withInputStreamWrapper(is)
          .withFileSize(totalSize).withFileSystem(fs).withFilePath(path).build();
        HFileBlock.FSReader hbr =
          new CorruptedFSReaderImpl(context, meta, TEST_UTIL.getConfiguration());
        HFileBlock b = hbr.readBlockData(0, -1, pread, false, true);
        b.sanityCheck();
        assertEquals(4936, b.getUncompressedSizeWithoutHeader());
        assertEquals(algo == GZ ? 2173 : 4936,
          b.getOnDiskSizeWithoutHeader() - b.totalChecksumBytes());
        // read data back from the hfile, exclude header and checksum
        ByteBuff bb = b.unpack(meta, hbr).getBufferWithoutHeader(); // read back data
        DataInputStream in =
          new DataInputStream(new ByteArrayInputStream(bb.array(), bb.arrayOffset(), bb.limit()));

        // assert that we encountered hbase checksum verification failures
        // but still used hdfs checksums and read data successfully.
        assertEquals(1, HFile.getAndResetChecksumFailuresCount());
        validateData(in);

        // A single instance of hbase checksum failure causes the reader to
        // switch off hbase checksum verification for the next 100 read
        // requests. Verify that this is correct.
        for (int i = 0; i < HFileBlock.CHECKSUM_VERIFICATION_NUM_IO_THRESHOLD + 1; i++) {
          b = hbr.readBlockData(0, -1, pread, false, true);
          assertTrue(b.getBufferReadOnly() instanceof SingleByteBuff);
          assertEquals(0, HFile.getAndResetChecksumFailuresCount());
        }
        // The next read should have hbase checksum verification reanabled,
        // we verify this by assertng that there was a hbase-checksum failure.
        b = hbr.readBlockData(0, -1, pread, false, true);
        assertTrue(b.getBufferReadOnly() instanceof SingleByteBuff);
        assertEquals(1, HFile.getAndResetChecksumFailuresCount());

        // Since the above encountered a checksum failure, we switch
        // back to not checking hbase checksums.
        b = hbr.readBlockData(0, -1, pread, false, true);
        assertTrue(b.getBufferReadOnly() instanceof SingleByteBuff);
        assertEquals(0, HFile.getAndResetChecksumFailuresCount());
        is.close();

        // Now, use a completely new reader. Switch off hbase checksums in
        // the configuration. In this case, we should not detect
        // any retries within hbase.
        Configuration conf = TEST_UTIL.getConfiguration();
        HFileSystem newfs = new HFileSystem(conf, false);
        assertEquals(false, newfs.useHBaseChecksum());
        is = new FSDataInputStreamWrapper(newfs, path);
        context = new ReaderContextBuilder().withInputStreamWrapper(is).withFileSize(totalSize)
          .withFileSystem(newfs).withFilePath(path).build();
        hbr = new CorruptedFSReaderImpl(context, meta, conf);
        b = hbr.readBlockData(0, -1, pread, false, true);
        is.close();
        b.sanityCheck();
        b = b.unpack(meta, hbr);
        assertEquals(4936, b.getUncompressedSizeWithoutHeader());
        assertEquals(algo == GZ ? 2173 : 4936,
          b.getOnDiskSizeWithoutHeader() - b.totalChecksumBytes());
        // read data back from the hfile, exclude header and checksum
        bb = b.getBufferWithoutHeader(); // read back data
        in =
          new DataInputStream(new ByteArrayInputStream(bb.array(), bb.arrayOffset(), bb.limit()));

        // assert that we did not encounter hbase checksum verification failures
        // but still used hdfs checksums and read data successfully.
        assertEquals(0, HFile.getAndResetChecksumFailuresCount());
        validateData(in);
      }
    }
  }
Introduce checksum failures and check that we can still read the data
hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestChecksum.java