@Test
  public void testSelectiveFlushWhenEnabled() throws IOException {
    // Set up the configuration, use new one to not conflict with minicluster in other tests
    Configuration conf = new HBaseTestingUtil().getConfiguration();
    conf.setLong(HConstants.HREGION_MEMSTORE_FLUSH_SIZE, 200 * 1024);
    conf.set(FlushPolicyFactory.HBASE_FLUSH_POLICY_KEY, FlushAllLargeStoresPolicy.class.getName());
    conf.setLong(FlushLargeStoresPolicy.HREGION_COLUMNFAMILY_FLUSH_SIZE_LOWER_BOUND_MIN, 40 * 1024);
    // Intialize the region
    HRegion region = initHRegion("testSelectiveFlushWithDataCompaction", conf);
    // 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));
        }
      }
    }

    long totalMemstoreSize = region.getMemStoreDataSize();

    // Find the smallest LSNs for edits wrt to each CF.
    long smallestSeqCF1 = region.getOldestSeqIdOfStore(FAMILY1);
    long smallestSeqCF2 = region.getOldestSeqIdOfStore(FAMILY2);
    long smallestSeqCF3 = region.getOldestSeqIdOfStore(FAMILY3);

    // Find the sizes of the memstores of each CF.
    MemStoreSize cf1MemstoreSize = region.getStore(FAMILY1).getMemStoreSize();
    MemStoreSize cf2MemstoreSize = region.getStore(FAMILY2).getMemStoreSize();
    MemStoreSize cf3MemstoreSize = region.getStore(FAMILY3).getMemStoreSize();

    // Get the overall smallest LSN in the region's memstores.
    long smallestSeqInRegionCurrentMemstore = AbstractTestFSWAL
      .getEarliestMemStoreSeqNum(getWAL(region), region.getRegionInfo().getEncodedNameAsBytes());

    // The overall smallest LSN in the region's memstores should be the same as
    // the LSN of the smallest edit in CF1
    assertEquals(smallestSeqCF1, smallestSeqInRegionCurrentMemstore);

    // Some other sanity checks.
    assertTrue(smallestSeqCF1 < smallestSeqCF2);
    assertTrue(smallestSeqCF2 < smallestSeqCF3);
    assertTrue(cf1MemstoreSize.getDataSize() > 0);
    assertTrue(cf2MemstoreSize.getDataSize() > 0);
    assertTrue(cf3MemstoreSize.getDataSize() > 0);

    // The total memstore size should be the same as the sum of the sizes of
    // memstores of CF1, CF2 and CF3.
    assertEquals(totalMemstoreSize, cf1MemstoreSize.getDataSize() + cf2MemstoreSize.getDataSize()
      + cf3MemstoreSize.getDataSize());

    // Flush!
    region.flush(false);

    // Will use these to check if anything changed.
    MemStoreSize oldCF2MemstoreSize = cf2MemstoreSize;
    MemStoreSize oldCF3MemstoreSize = cf3MemstoreSize;

    // Recalculate everything
    cf1MemstoreSize = region.getStore(FAMILY1).getMemStoreSize();
    cf2MemstoreSize = region.getStore(FAMILY2).getMemStoreSize();
    cf3MemstoreSize = region.getStore(FAMILY3).getMemStoreSize();
    totalMemstoreSize = region.getMemStoreDataSize();
    smallestSeqInRegionCurrentMemstore = AbstractTestFSWAL.getEarliestMemStoreSeqNum(getWAL(region),
      region.getRegionInfo().getEncodedNameAsBytes());

    // We should have cleared out only CF1, since we chose the flush thresholds
    // and number of puts accordingly.
    assertEquals(0, cf1MemstoreSize.getDataSize());
    assertEquals(MutableSegment.DEEP_OVERHEAD, cf1MemstoreSize.getHeapSize());
    // Nothing should have happened to CF2, ...
    assertEquals(cf2MemstoreSize, oldCF2MemstoreSize);
    // ... or CF3
    assertEquals(cf3MemstoreSize, oldCF3MemstoreSize);
    // Now the smallest LSN in the region should be the same as the smallest
    // LSN in the memstore of CF2.
    assertEquals(smallestSeqInRegionCurrentMemstore, smallestSeqCF2);
    // Of course, this should hold too.
    assertEquals(totalMemstoreSize, cf2MemstoreSize.getDataSize() + cf3MemstoreSize.getDataSize());

    // Now add more puts (mostly for CF2), so that we only flush CF2 this time.
    for (int i = 1200; i < 2400; i++) {
      region.put(createPut(2, i));

      // Add only 100 puts for CF3
      if (i - 1200 < 100) {
        region.put(createPut(3, i));
      }
    }

    // How much does the CF3 memstore occupy? Will be used later.
    oldCF3MemstoreSize = region.getStore(FAMILY3).getMemStoreSize();

    // Flush again
    region.flush(false);

    // Recalculate everything
    cf1MemstoreSize = region.getStore(FAMILY1).getMemStoreSize();
    cf2MemstoreSize = region.getStore(FAMILY2).getMemStoreSize();
    cf3MemstoreSize = region.getStore(FAMILY3).getMemStoreSize();
    totalMemstoreSize = region.getMemStoreDataSize();
    smallestSeqInRegionCurrentMemstore = AbstractTestFSWAL.getEarliestMemStoreSeqNum(getWAL(region),
      region.getRegionInfo().getEncodedNameAsBytes());

    // CF1 and CF2, both should be absent.
    assertEquals(0, cf1MemstoreSize.getDataSize());
    assertEquals(MutableSegment.DEEP_OVERHEAD, cf1MemstoreSize.getHeapSize());
    assertEquals(0, cf2MemstoreSize.getDataSize());
    assertEquals(MutableSegment.DEEP_OVERHEAD, cf2MemstoreSize.getHeapSize());
    // CF3 shouldn't have been touched.
    assertEquals(cf3MemstoreSize, oldCF3MemstoreSize);
    assertEquals(totalMemstoreSize, cf3MemstoreSize.getDataSize());
    assertEquals(smallestSeqInRegionCurrentMemstore, smallestSeqCF3);

    // 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.

    // Clearing the existing memstores.
    region.flush(true);

    // 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));
    }

    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.
    assertEquals(0, region.getMemStoreDataSize());
    HBaseTestingUtil.closeRegionAndWAL(region);
  }
This test verifies the correctness of the Per Column Family flushing strategy
hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestPerColumnFamilyFlush.java