@Test
  public void testSelectiveFlushWithIndexCompaction() throws IOException {
    /*------------------------------------------------------------------------------*/
    /* SETUP */
    // 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);
    conf.setDouble(CompactingMemStore.IN_MEMORY_FLUSH_THRESHOLD_FACTOR_KEY, 0.5);
    // set memstore to index-compaction
    conf.set(CompactingMemStore.COMPACTING_MEMSTORE_TYPE_KEY,
      String.valueOf(MemoryCompactionPolicy.BASIC));

    // Initialize the region
    HRegion region = initHRegion("testSelectiveFlushWithIndexCompaction", conf);
    verifyInMemoryFlushSize(region);
    /*------------------------------------------------------------------------------*/
    /* PHASE I - insertions */
    // Add 1200 entries for CF1, 100 for CF2 and 50 for CF3
    for (int i = 1; i <= 1200; i++) {
      region.put(createPut(1, i)); // compacted memstore
      if (i <= 100) {
        region.put(createPut(2, i));
        if (i <= 50) {
          region.put(createDoublePut(3, i)); // subject for in-memory compaction
        }
      }
    }
    // 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));
    }

    /*------------------------------------------------------------------------------*/
    /*------------------------------------------------------------------------------*/
    /* PHASE I - collect sizes */
    long totalMemstoreSizePhaseI = region.getMemStoreDataSize();
    // Find the smallest LSNs for edits wrt to each CF.
    long smallestSeqCF1PhaseI = region.getOldestSeqIdOfStore(FAMILY1);
    long smallestSeqCF2PhaseI = region.getOldestSeqIdOfStore(FAMILY2);
    long smallestSeqCF3PhaseI = region.getOldestSeqIdOfStore(FAMILY3);
    // 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();
    // Get the overall smallest LSN in the region's memstores.
    long smallestSeqInRegionCurrentMemstorePhaseI = AbstractTestFSWAL
      .getEarliestMemStoreSeqNum(getWAL(region), region.getRegionInfo().getEncodedNameAsBytes());

    /*------------------------------------------------------------------------------*/
    /* PHASE I - validation */
    // The overall smallest LSN in the region's memstores should be the same as
    // the LSN of the smallest edit in CF1
    assertEquals(smallestSeqCF1PhaseI, smallestSeqInRegionCurrentMemstorePhaseI);
    // Some other sanity checks.
    assertTrue(smallestSeqCF1PhaseI < smallestSeqCF2PhaseI);
    assertTrue(smallestSeqCF2PhaseI < smallestSeqCF3PhaseI);
    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.
    assertEquals(totalMemstoreSizePhaseI, cf1MemstoreSizePhaseI.getDataSize()
      + cf2MemstoreSizePhaseI.getDataSize() + cf3MemstoreSizePhaseI.getDataSize());

    /*------------------------------------------------------------------------------*/
    /* PHASE I - Flush */
    // First Flush in Test!!!!!!!!!!!!!!!!!!!!!!
    // CF1, CF2, CF3, all together they are above the flush size lower bound.
    // Since CF1 and CF3 are compacting, CF2 is going to be flushed to disk.
    // CF1 and CF3 - flushed to memory and flatten explicitly
    region.flush(false);
    CompactingMemStore cms1 = (CompactingMemStore) ((HStore) region.getStore(FAMILY1)).memstore;
    CompactingMemStore cms3 = (CompactingMemStore) ((HStore) region.getStore(FAMILY3)).memstore;
    cms1.flushInMemory();
    cms3.flushInMemory();

    // CF3/CF1 should be merged so wait here to be sure the compaction is done
    while (
      ((CompactingMemStore) ((HStore) region.getStore(FAMILY1)).memstore)
        .isMemStoreFlushingInMemory()
    ) {
      Threads.sleep(10);
    }
    while (
      ((CompactingMemStore) ((HStore) region.getStore(FAMILY3)).memstore)
        .isMemStoreFlushingInMemory()
    ) {
      Threads.sleep(10);
    }

    /*------------------------------------------------------------------------------*/
    /*------------------------------------------------------------------------------*/
    /* PHASE II - collect sizes */
    // Recalculate everything
    MemStoreSize cf1MemstoreSizePhaseII = region.getStore(FAMILY1).getMemStoreSize();
    MemStoreSize cf2MemstoreSizePhaseII = region.getStore(FAMILY2).getMemStoreSize();
    MemStoreSize cf3MemstoreSizePhaseII = region.getStore(FAMILY3).getMemStoreSize();
    long smallestSeqInRegionCurrentMemstorePhaseII = AbstractTestFSWAL
      .getEarliestMemStoreSeqNum(getWAL(region), region.getRegionInfo().getEncodedNameAsBytes());
    // Find the smallest LSNs for edits wrt to each CF.
    long smallestSeqCF3PhaseII = region.getOldestSeqIdOfStore(FAMILY3);
    long totalMemstoreSizePhaseII = region.getMemStoreDataSize();

    /*------------------------------------------------------------------------------*/
    /* PHASE II - validation */
    // CF1 was flushed to memory, should be flattened and take less space
    assertEquals(cf1MemstoreSizePhaseII.getDataSize(), cf1MemstoreSizePhaseI.getDataSize());
    assertTrue(cf1MemstoreSizePhaseII.getHeapSize() < cf1MemstoreSizePhaseI.getHeapSize());
    // CF2 should become empty
    assertEquals(0, cf2MemstoreSizePhaseII.getDataSize());
    assertEquals(MutableSegment.DEEP_OVERHEAD, cf2MemstoreSizePhaseII.getHeapSize());
    // verify that CF3 was flushed to memory and was not compacted (this is an approximation check)
    // if compacted CF# should be at least twice less because its every key was duplicated
    assertEquals(cf3MemstoreSizePhaseII.getDataSize(), cf3MemstoreSizePhaseI.getDataSize());
    assertTrue(cf3MemstoreSizePhaseI.getHeapSize() / 2 < cf3MemstoreSizePhaseII.getHeapSize());

    // Now the smallest LSN in the region should be the same as the smallest
    // LSN in the memstore of CF1.
    assertEquals(smallestSeqInRegionCurrentMemstorePhaseII, smallestSeqCF1PhaseI);
    // The total memstore size should be the same as the sum of the sizes of
    // memstores of CF1, CF2 and CF3. Counting the empty active segments in CF1/2/3 and pipeline
    // items in CF1/2
    assertEquals(totalMemstoreSizePhaseII, cf1MemstoreSizePhaseII.getDataSize()
      + cf2MemstoreSizePhaseII.getDataSize() + cf3MemstoreSizePhaseII.getDataSize());

    /*------------------------------------------------------------------------------*/
    /*------------------------------------------------------------------------------*/
    /* PHASE III - insertions */
    // Now add more puts for CF1, so that we also flush CF1 to disk instead of
    // memory in next flush. This is causing the CF! to be flushed to memory twice.
    for (int i = 1200; i < 8000; i++) {
      region.put(createPut(1, i));
    }

    // CF1 should be flatten and merged so wait here to be sure the compaction is done
    while (
      ((CompactingMemStore) ((HStore) region.getStore(FAMILY1)).memstore)
        .isMemStoreFlushingInMemory()
    ) {
      Threads.sleep(10);
    }

    /*------------------------------------------------------------------------------*/
    /* PHASE III - collect sizes */
    // How much does the CF1 memstore occupy now? Will be used later.
    MemStoreSize cf1MemstoreSizePhaseIII = region.getStore(FAMILY1).getMemStoreSize();
    long totalMemstoreSizePhaseIII = region.getMemStoreDataSize();

    /*------------------------------------------------------------------------------*/
    /* PHASE III - validation */
    // The total memstore size should be the same as the sum of the sizes of
    // memstores of CF1, CF2 and CF3. Counting the empty active segments in CF1/2/3 and pipeline
    // items in CF1/2
    assertEquals(totalMemstoreSizePhaseIII, cf1MemstoreSizePhaseIII.getDataSize()
      + cf2MemstoreSizePhaseII.getDataSize() + cf3MemstoreSizePhaseII.getDataSize());

    /*------------------------------------------------------------------------------*/
    /* PHASE III - Flush */
    // Second Flush in Test!!!!!!!!!!!!!!!!!!!!!!
    // CF1 is flushed to disk, but not entirely emptied.
    // CF2 was and remained empty, same way nothing happens to CF3
    region.flush(false);

    /*------------------------------------------------------------------------------*/
    /*------------------------------------------------------------------------------*/
    /* PHASE IV - collect sizes */
    // Recalculate everything
    MemStoreSize cf1MemstoreSizePhaseIV = region.getStore(FAMILY1).getMemStoreSize();
    MemStoreSize cf2MemstoreSizePhaseIV = region.getStore(FAMILY2).getMemStoreSize();
    MemStoreSize cf3MemstoreSizePhaseIV = region.getStore(FAMILY3).getMemStoreSize();
    long smallestSeqInRegionCurrentMemstorePhaseIV = AbstractTestFSWAL
      .getEarliestMemStoreSeqNum(getWAL(region), region.getRegionInfo().getEncodedNameAsBytes());
    long smallestSeqCF3PhaseIV = region.getOldestSeqIdOfStore(FAMILY3);

    /*------------------------------------------------------------------------------*/
    /* PHASE IV - validation */
    // CF1's biggest pipeline component (inserted before first flush) should be flushed to disk
    // CF2 should remain empty
    assertTrue(cf1MemstoreSizePhaseIII.getDataSize() > cf1MemstoreSizePhaseIV.getDataSize());
    assertEquals(0, cf2MemstoreSizePhaseIV.getDataSize());
    assertEquals(MutableSegment.DEEP_OVERHEAD, cf2MemstoreSizePhaseIV.getHeapSize());
    // CF3 shouldn't have been touched.
    assertEquals(cf3MemstoreSizePhaseIV, cf3MemstoreSizePhaseII);
    // the smallest LSN of CF3 shouldn't change
    assertEquals(smallestSeqCF3PhaseII, smallestSeqCF3PhaseIV);
    // CF3 should be bottleneck for WAL
    assertEquals(smallestSeqInRegionCurrentMemstorePhaseIV, smallestSeqCF3PhaseIV);

    /*------------------------------------------------------------------------------*/
    /* PHASE IV - Flush */
    // Third Flush in Test!!!!!!!!!!!!!!!!!!!!!!
    // Force flush to disk on all memstores (flush parameter true).
    // CF1/CF3 all flushed to disk. Note that active sets of CF1 and CF3 are empty
    region.flush(true);

    /*------------------------------------------------------------------------------*/
    /*------------------------------------------------------------------------------*/
    /* PHASE V - collect sizes */
    // Recalculate everything
    MemStoreSize cf1MemstoreSizePhaseV = region.getStore(FAMILY1).getMemStoreSize();
    MemStoreSize cf2MemstoreSizePhaseV = region.getStore(FAMILY2).getMemStoreSize();
    MemStoreSize cf3MemstoreSizePhaseV = region.getStore(FAMILY3).getMemStoreSize();
    long smallestSeqInRegionCurrentMemstorePhaseV = AbstractTestFSWAL
      .getEarliestMemStoreSeqNum(getWAL(region), region.getRegionInfo().getEncodedNameAsBytes());
    long totalMemstoreSizePhaseV = region.getMemStoreDataSize();

    /*------------------------------------------------------------------------------*/
    /* PHASE V - validation */
    assertEquals(0, cf1MemstoreSizePhaseV.getDataSize());
    assertEquals(MutableSegment.DEEP_OVERHEAD, cf1MemstoreSizePhaseV.getHeapSize());
    assertEquals(0, cf2MemstoreSizePhaseV.getDataSize());
    assertEquals(MutableSegment.DEEP_OVERHEAD, cf2MemstoreSizePhaseV.getHeapSize());
    assertEquals(0, cf3MemstoreSizePhaseV.getDataSize());
    assertEquals(MutableSegment.DEEP_OVERHEAD, cf3MemstoreSizePhaseV.getHeapSize());
    // The total memstores size should be empty
    assertEquals(0, totalMemstoreSizePhaseV);
    // Because there is nothing in any memstore the WAL's LSN should be -1
    assertEquals(HConstants.NO_SEQNUM, smallestSeqInRegionCurrentMemstorePhaseV);

    // What happens when we hit the memstore limit, but we are not able to find
    // any Column Family above the threshold?
    // In that case, we should flush all the CFs.

    /*------------------------------------------------------------------------------*/
    /*------------------------------------------------------------------------------*/
    /* PHASE VI - insertions */
    // The memstore limit is 200*1024 and the column family flush threshold is
    // around 50*1024. We try to just hit the memstore limit with each CF's
    // memstore being below the CF flush threshold.
    for (int i = 1; i <= 300; i++) {
      region.put(createPut(1, i));
      region.put(createPut(2, i));
      region.put(createPut(3, i));
      region.put(createPut(4, i));
      region.put(createPut(5, i));
    }

    MemStoreSize cf1ActiveSizePhaseVI = region.getStore(FAMILY1).getMemStoreSize();
    MemStoreSize cf3ActiveSizePhaseVI = region.getStore(FAMILY3).getMemStoreSize();
    MemStoreSize cf5ActiveSizePhaseVI = region.getStore(FAMILIES[4]).getMemStoreSize();

    /*------------------------------------------------------------------------------*/
    /* PHASE VI - Flush */
    // Fourth Flush in Test!!!!!!!!!!!!!!!!!!!!!!
    // None among compacting memstores was flushed to memory due to previous puts.
    // But is going to be moved to pipeline and flatten due to the flush.
    region.flush(false);
    // Since we won't find any CF above the threshold, and hence no specific
    // store to flush, we should flush all the memstores
    // Also compacted memstores are flushed to disk, but not entirely emptied
    MemStoreSize cf1ActiveSizePhaseVII = region.getStore(FAMILY1).getMemStoreSize();
    MemStoreSize cf3ActiveSizePhaseVII = region.getStore(FAMILY3).getMemStoreSize();
    MemStoreSize cf5ActiveSizePhaseVII = region.getStore(FAMILIES[4]).getMemStoreSize();

    assertTrue(cf1ActiveSizePhaseVII.getDataSize() < cf1ActiveSizePhaseVI.getDataSize());
    assertTrue(cf3ActiveSizePhaseVII.getDataSize() < cf3ActiveSizePhaseVI.getDataSize());
    assertTrue(cf5ActiveSizePhaseVII.getDataSize() < cf5ActiveSizePhaseVI.getDataSize());

    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