@Test
  public void testSelectiveFlushWithBasicAndMerge() 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);
    conf.setDouble(CompactingMemStore.IN_MEMORY_FLUSH_THRESHOLD_FACTOR_KEY, 0.8);
    // set memstore to do index compaction with merge
    conf.set(CompactingMemStore.COMPACTING_MEMSTORE_TYPE_KEY,
      String.valueOf(MemoryCompactionPolicy.BASIC));
    // length of pipeline that requires merge
    conf.setInt(MemStoreCompactionStrategy.COMPACTING_MEMSTORE_THRESHOLD_KEY, 1);

    // Intialize the HRegion
    HRegion region = initHRegion("testSelectiveFlushWithBasicAndMerge", conf);
    verifyInMemoryFlushSize(region);
    // Add 1200 entries for CF1 (CompactingMemStore), 100 for CF2 (DefaultMemStore) 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 put more entries to CF2
    for (int i = 100; i < 2000; i++) {
      region.put(createPut(2, i));
    }

    long totalMemstoreSize = region.getMemStoreDataSize();

    // test in-memory flashing into CAM here
    ((CompactingMemStore) ((HStore) region.getStore(FAMILY1)).memstore)
      .setIndexType(CompactingMemStore.IndexType.ARRAY_MAP);
    ((CompactingMemStore) ((HStore) region.getStore(FAMILY3)).memstore)
      .setIndexType(CompactingMemStore.IndexType.ARRAY_MAP);

    // 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.
    assertEquals(totalMemstoreSize, cf1MemstoreSizePhaseI.getDataSize()
      + cf2MemstoreSizePhaseI.getDataSize() + cf3MemstoreSizePhaseI.getDataSize());

    // Initiate in-memory Flush!
    ((CompactingMemStore) ((HStore) region.getStore(FAMILY1)).memstore).flushInMemory();
    ((CompactingMemStore) ((HStore) region.getStore(FAMILY3)).memstore).flushInMemory();
    // CF1 and CF3 should be flatten and merged so wait here to be sure the merge is done
    while (
      ((CompactingMemStore) ((HStore) region.getStore(FAMILY1)).memstore)
        .isMemStoreFlushingInMemory()
    ) {
      Threads.sleep(10);
    }
    while (
      ((CompactingMemStore) ((HStore) region.getStore(FAMILY3)).memstore)
        .isMemStoreFlushingInMemory()
    ) {
      Threads.sleep(10);
    }

    // Flush-to-disk! CF2 only should be flushed
    region.flush(false);

    MemStoreSize cf1MemstoreSizePhaseII = region.getStore(FAMILY1).getMemStoreSize();
    MemStoreSize cf2MemstoreSizePhaseII = region.getStore(FAMILY2).getMemStoreSize();
    MemStoreSize cf3MemstoreSizePhaseII = region.getStore(FAMILY3).getMemStoreSize();

    // CF1 should be flushed in memory and just flattened, so CF1 heap overhead should be smaller
    assertTrue(cf1MemstoreSizePhaseI.getHeapSize() > cf1MemstoreSizePhaseII.getHeapSize());
    // CF1 should be flushed in memory and just flattened, so CF1 data size should remain the same
    assertEquals(cf1MemstoreSizePhaseI.getDataSize(), cf1MemstoreSizePhaseII.getDataSize());
    // CF2 should have been cleared
    assertEquals(0, cf2MemstoreSizePhaseII.getDataSize());

    // Add the same amount of entries to see the merging
    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));
    }

    MemStoreSize cf1MemstoreSizePhaseIII = region.getStore(FAMILY1).getMemStoreSize();

    // Flush in memory!
    ((CompactingMemStore) ((HStore) region.getStore(FAMILY1)).memstore).flushInMemory();
    ((CompactingMemStore) ((HStore) region.getStore(FAMILY3)).memstore).flushInMemory();
    // CF1 and CF3 should be merged so wait here to be sure the merge is done
    while (
      ((CompactingMemStore) ((HStore) region.getStore(FAMILY1)).memstore)
        .isMemStoreFlushingInMemory()
    ) {
      Threads.sleep(10);
    }
    while (
      ((CompactingMemStore) ((HStore) region.getStore(FAMILY3)).memstore)
        .isMemStoreFlushingInMemory()
    ) {
      Threads.sleep(10);
    }
    region.flush(false);

    MemStoreSize cf1MemstoreSizePhaseIV = region.getStore(FAMILY1).getMemStoreSize();
    MemStoreSize cf2MemstoreSizePhaseIV = region.getStore(FAMILY2).getMemStoreSize();

    assertEquals(2 * cf1MemstoreSizePhaseI.getDataSize(), cf1MemstoreSizePhaseIV.getDataSize());
    // the decrease in the heap size due to usage of CellArrayMap instead of CSLM
    // should be the same in flattening and in merge (first and second in-memory-flush)
    // but in phase 1 we do not yet have immutable segment
    assertEquals(cf1MemstoreSizePhaseI.getHeapSize() - cf1MemstoreSizePhaseII.getHeapSize(),
      cf1MemstoreSizePhaseIII.getHeapSize() - cf1MemstoreSizePhaseIV.getHeapSize()
        - CellArrayImmutableSegment.DEEP_OVERHEAD_CAM);
    assertEquals(3, // active, one in pipeline, snapshot
      ((CompactingMemStore) ((HStore) region.getStore(FAMILY1)).memstore).getSegments().size());
    // CF2 should have been cleared
    assertEquals("\n<<< DEBUG: The data--heap sizes of stores before/after first flushes,"
      + " CF1: " + cf1MemstoreSizePhaseI.getDataSize() + "/" + cf1MemstoreSizePhaseII.getDataSize()
      + "--" + cf1MemstoreSizePhaseI.getHeapSize() + "/" + cf1MemstoreSizePhaseII.getHeapSize()
      + ", CF2: " + cf2MemstoreSizePhaseI.getDataSize() + "/" + cf2MemstoreSizePhaseII.getDataSize()
      + "--" + cf2MemstoreSizePhaseI.getHeapSize() + "/" + cf2MemstoreSizePhaseII.getHeapSize()
      + ", CF3: " + cf3MemstoreSizePhaseI.getDataSize() + "/" + cf3MemstoreSizePhaseII.getDataSize()
      + "--" + cf3MemstoreSizePhaseI.getHeapSize() + "/" + cf3MemstoreSizePhaseII.getHeapSize()
      + "\n<<< AND before/after second flushes " + " CF1: " + cf1MemstoreSizePhaseIII.getDataSize()
      + "/" + cf1MemstoreSizePhaseIV.getDataSize() + "--" + cf1MemstoreSizePhaseIII.getHeapSize()
      + "/" + cf1MemstoreSizePhaseIV.getHeapSize() + "\n", 0, cf2MemstoreSizePhaseIV.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