@Test
  public void testFlushAndCompactionwithCombinations() throws Exception {
    TableName tableName = TableName.valueOf(TEST_NAME.getMethodName());
    byte[] fam = Bytes.toBytes("info");
    byte[] row = Bytes.toBytes("rowa");
    // column names
    byte[] qual = Bytes.toBytes("qual");

    byte[] row1 = Bytes.toBytes("rowb");

    byte[] row2 = Bytes.toBytes("rowc");
    byte[] rowd = Bytes.toBytes("rowd");
    byte[] rowe = Bytes.toBytes("rowe");
    Table table = null;
    for (DataBlockEncoding encoding : DataBlockEncoding.values()) {
      TableDescriptor tableDescriptor = TableDescriptorBuilder.newBuilder(tableName)
        .setColumnFamily(ColumnFamilyDescriptorBuilder.newBuilder(fam).setBlockCacheEnabled(true)
          .setDataBlockEncoding(encoding).build())
        .build();
      Admin admin = TEST_UTIL.getAdmin();
      admin.createTable(tableDescriptor);
      try {
        table = TEST_UTIL.getConnection().getTable(tableName);
        Put put = new Put(row);
        byte[] value = Bytes.toBytes("value");
        put.addColumn(fam, qual, HConstants.LATEST_TIMESTAMP, value);
        int bigTagLen = Short.MAX_VALUE - 5;
        put.setAttribute("visibility", new byte[bigTagLen]);
        table.put(put);
        Put put1 = new Put(row1);
        byte[] value1 = Bytes.toBytes("1000dfsdf");
        put1.addColumn(fam, qual, HConstants.LATEST_TIMESTAMP, value1);
        table.put(put1);
        admin.flush(tableName);
        // We are lacking an API for confirming flush request compaction.
        // Just sleep for a short time. We won't be able to confirm flush
        // completion but the test won't hang now or in the future if
        // default compaction policy causes compaction between flush and
        // when we go to confirm it.
        Thread.sleep(1000);

        put1 = new Put(row2);
        value1 = Bytes.toBytes("1000dfsdf");
        put1.addColumn(fam, qual, HConstants.LATEST_TIMESTAMP, value1);
        table.put(put1);
        admin.flush(tableName);
        Thread.sleep(1000);

        Put put2 = new Put(rowd);
        byte[] value2 = Bytes.toBytes("1000dfsdf");
        put2.addColumn(fam, qual, HConstants.LATEST_TIMESTAMP, value2);
        table.put(put2);
        put2 = new Put(rowe);
        value2 = Bytes.toBytes("1000dfsddfdf");
        put2.addColumn(fam, qual, HConstants.LATEST_TIMESTAMP, value2);
        put.setAttribute("visibility", Bytes.toBytes("ram"));
        table.put(put2);
        admin.flush(tableName);
        Thread.sleep(1000);

        TestCoprocessorForTags.checkTagPresence = true;
        Scan s = new Scan().withStartRow(row);
        s.setCaching(1);
        ResultScanner scanner = table.getScanner(s);
        try {
          Result next = null;
          while ((next = scanner.next()) != null) {
            CellScanner cellScanner = next.cellScanner();
            cellScanner.advance();
            Cell current = cellScanner.current();
            if (CellUtil.matchingRows(current, row)) {
              assertEquals(1, TestCoprocessorForTags.tags.size());
              Tag tag = TestCoprocessorForTags.tags.get(0);
              assertEquals(bigTagLen, tag.getValueLength());
            } else {
              assertEquals(0, TestCoprocessorForTags.tags.size());
            }
          }
        } finally {
          if (scanner != null) {
            scanner.close();
          }
          TestCoprocessorForTags.checkTagPresence = false;
        }
        while (admin.getCompactionState(tableName) != CompactionState.NONE) {
          Thread.sleep(10);
        }
        TestCoprocessorForTags.checkTagPresence = true;
        scanner = table.getScanner(s);
        try {
          Result next = null;
          while ((next = scanner.next()) != null) {
            CellScanner cellScanner = next.cellScanner();
            cellScanner.advance();
            Cell current = cellScanner.current();
            if (CellUtil.matchingRows(current, row)) {
              assertEquals(1, TestCoprocessorForTags.tags.size());
              Tag tag = TestCoprocessorForTags.tags.get(0);
              assertEquals(bigTagLen, tag.getValueLength());
            } else {
              assertEquals(0, TestCoprocessorForTags.tags.size());
            }
          }
        } finally {
          if (scanner != null) {
            scanner.close();
          }
          TestCoprocessorForTags.checkTagPresence = false;
        }
      } finally {
        if (table != null) {
          table.close();
        }
        // delete the table
        admin.disableTable(tableName);
        admin.deleteTable(tableName);
      }
    }
  }
Test that we can do reverse scans when writing tags and using DataBlockEncoding. Fails with an
exception for PREFIX, DIFF, and FAST_DIFF
hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestTags.java