@Test
  public void testSelectiveFlushAndWALinDataCompaction() throws IOException {
    // Set up the configuration
    conf.setLong(HConstants.HREGION_MEMSTORE_FLUSH_SIZE, 300 * 1024);
    conf.setLong(FlushLargeStoresPolicy.HREGION_COLUMNFAMILY_FLUSH_SIZE_LOWER_BOUND_MIN, 75 * 1024);
    // set memstore to do data compaction and not to use the speculative scan
    conf.set(CompactingMemStore.COMPACTING_MEMSTORE_TYPE_KEY,
      String.valueOf(MemoryCompactionPolicy.EAGER));

    // Intialize the HRegion
    HRegion region = initHRegion("testSelectiveFlushAndWALinDataCompaction", conf);
    verifyInMemoryFlushSize(region);
    // Add 1200 entries for CF1, 100 for CF2 and 50 for CF3
    for (int i = 1; i <= 1200; i++) {
      region.put(createPut(1, i));
      if (i <= 100) {
        region.put(createPut(2, i));
        if (i <= 50) {
          region.put(createPut(3, i));
        }
      }
    }
    // Now add more puts for CF2, so that we only flush CF2 to disk
    for (int i = 100; i < 2000; i++) {
      region.put(createPut(2, i));
    }

    // in this test check the non-composite snapshot - flashing only tail of the pipeline
    ((CompactingMemStore) ((HStore) region.getStore(FAMILY1)).memstore).setCompositeSnapshot(false);
    ((CompactingMemStore) ((HStore) region.getStore(FAMILY3)).memstore).setCompositeSnapshot(false);

    long totalMemstoreSize = region.getMemStoreDataSize();

    // Find the sizes of the memstores of each CF.
    MemStoreSize cf1MemstoreSizePhaseI = region.getStore(FAMILY1).getMemStoreSize();
    MemStoreSize cf2MemstoreSizePhaseI = region.getStore(FAMILY2).getMemStoreSize();
    MemStoreSize cf3MemstoreSizePhaseI = region.getStore(FAMILY3).getMemStoreSize();

    // Some other sanity checks.
    assertTrue(cf1MemstoreSizePhaseI.getDataSize() > 0);
    assertTrue(cf2MemstoreSizePhaseI.getDataSize() > 0);
    assertTrue(cf3MemstoreSizePhaseI.getDataSize() > 0);

    // The total memstore size should be the same as the sum of the sizes of
    // memstores of CF1, CF2 and CF3.
    String msg = "totalMemstoreSize=" + totalMemstoreSize + " DefaultMemStore.DEEP_OVERHEAD="
      + DefaultMemStore.DEEP_OVERHEAD + " cf1MemstoreSizePhaseI=" + cf1MemstoreSizePhaseI
      + " cf2MemstoreSizePhaseI=" + cf2MemstoreSizePhaseI + " cf3MemstoreSizePhaseI="
      + cf3MemstoreSizePhaseI;
    assertEquals(msg, totalMemstoreSize, cf1MemstoreSizePhaseI.getDataSize()
      + cf2MemstoreSizePhaseI.getDataSize() + cf3MemstoreSizePhaseI.getDataSize());

    // Flush!
    CompactingMemStore cms1 = (CompactingMemStore) ((HStore) region.getStore(FAMILY1)).memstore;
    CompactingMemStore cms3 = (CompactingMemStore) ((HStore) region.getStore(FAMILY3)).memstore;
    cms1.flushInMemory();
    cms3.flushInMemory();
    region.flush(false);

    MemStoreSize cf2MemstoreSizePhaseII = region.getStore(FAMILY2).getMemStoreSize();

    long smallestSeqInRegionCurrentMemstorePhaseII = AbstractTestFSWAL
      .getEarliestMemStoreSeqNum(region.getWAL(), region.getRegionInfo().getEncodedNameAsBytes());
    long smallestSeqCF1PhaseII = region.getOldestSeqIdOfStore(FAMILY1);
    long smallestSeqCF2PhaseII = region.getOldestSeqIdOfStore(FAMILY2);
    long smallestSeqCF3PhaseII = region.getOldestSeqIdOfStore(FAMILY3);

    // CF2 should have been cleared
    assertEquals(0, cf2MemstoreSizePhaseII.getDataSize());
    assertEquals(MutableSegment.DEEP_OVERHEAD, cf2MemstoreSizePhaseII.getHeapSize());

    String s = "\n\n----------------------------------\n"
      + "Upon initial insert and flush, LSN of CF1 is:" + smallestSeqCF1PhaseII + ". LSN of CF2 is:"
      + smallestSeqCF2PhaseII + ". LSN of CF3 is:" + smallestSeqCF3PhaseII
      + ", smallestSeqInRegionCurrentMemstore:" + smallestSeqInRegionCurrentMemstorePhaseII + "\n";

    // Add same entries to compact them later
    for (int i = 1; i <= 1200; i++) {
      region.put(createPut(1, i));
      if (i <= 100) {
        region.put(createPut(2, i));
        if (i <= 50) {
          region.put(createPut(3, i));
        }
      }
    }
    // Now add more puts for CF2, so that we only flush CF2 to disk
    for (int i = 100; i < 2000; i++) {
      region.put(createPut(2, i));
    }

    long smallestSeqInRegionCurrentMemstorePhaseIII = AbstractTestFSWAL
      .getEarliestMemStoreSeqNum(region.getWAL(), region.getRegionInfo().getEncodedNameAsBytes());
    long smallestSeqCF1PhaseIII = region.getOldestSeqIdOfStore(FAMILY1);
    long smallestSeqCF2PhaseIII = region.getOldestSeqIdOfStore(FAMILY2);
    long smallestSeqCF3PhaseIII = region.getOldestSeqIdOfStore(FAMILY3);

    s = s + "The smallest sequence in region WAL is: " + smallestSeqInRegionCurrentMemstorePhaseIII
      + ", the smallest sequence in CF1:" + smallestSeqCF1PhaseIII + ", "
      + "the smallest sequence in CF2:" + smallestSeqCF2PhaseIII + ", the smallest sequence in CF3:"
      + smallestSeqCF3PhaseIII + "\n";

    // Flush!
    cms1 = (CompactingMemStore) ((HStore) region.getStore(FAMILY1)).memstore;
    cms3 = (CompactingMemStore) ((HStore) region.getStore(FAMILY3)).memstore;
    cms1.flushInMemory();
    cms3.flushInMemory();
    region.flush(false);

    long smallestSeqInRegionCurrentMemstorePhaseIV = AbstractTestFSWAL
      .getEarliestMemStoreSeqNum(region.getWAL(), region.getRegionInfo().getEncodedNameAsBytes());
    long smallestSeqCF1PhaseIV = region.getOldestSeqIdOfStore(FAMILY1);
    long smallestSeqCF2PhaseIV = region.getOldestSeqIdOfStore(FAMILY2);
    long smallestSeqCF3PhaseIV = region.getOldestSeqIdOfStore(FAMILY3);

    s = s + "The smallest sequence in region WAL is: " + smallestSeqInRegionCurrentMemstorePhaseIV
      + ", the smallest sequence in CF1:" + smallestSeqCF1PhaseIV + ", "
      + "the smallest sequence in CF2:" + smallestSeqCF2PhaseIV + ", the smallest sequence in CF3:"
      + smallestSeqCF3PhaseIV + "\n";

    // now check that the LSN of the entire WAL, of CF1 and of CF3 has progressed due to compaction
    assertTrue(s,
      smallestSeqInRegionCurrentMemstorePhaseIV > smallestSeqInRegionCurrentMemstorePhaseIII);
    assertTrue(smallestSeqCF1PhaseIV > smallestSeqCF1PhaseIII);
    assertTrue(smallestSeqCF3PhaseIV > smallestSeqCF3PhaseIII);

    HBaseTestingUtil.closeRegionAndWAL(region);
  }
This test verifies the correctness of the Per Column Family flushing strategy when part of the
memstores are compacted memstores
hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestWalAndCompactingMemStoreFlush.java