@Test
  public void testCompareStoreFileCount() throws Exception {
    long memstoreFlushSize = 1024L * 1024;
    Configuration conf = TEST_UTIL.getConfiguration();
    conf.setLong(HConstants.HREGION_MEMSTORE_FLUSH_SIZE, memstoreFlushSize);
    conf.set(FlushPolicyFactory.HBASE_FLUSH_POLICY_KEY, FlushAllStoresPolicy.class.getName());
    conf.setInt(HStore.BLOCKING_STOREFILES_KEY, 10000);
    conf.set(HConstants.HBASE_REGION_SPLIT_POLICY_KEY,
      ConstantSizeRegionSplitPolicy.class.getName());

    TableDescriptor tableDescriptor = TableDescriptorBuilder.newBuilder(TABLENAME)
      .setCompactionEnabled(false).setColumnFamily(ColumnFamilyDescriptorBuilder.of(FAMILY1))
      .setColumnFamily(ColumnFamilyDescriptorBuilder.of(FAMILY2))
      .setColumnFamily(ColumnFamilyDescriptorBuilder.of(FAMILY3)).build();

    LOG.info("==============Test with selective flush disabled===============");
    int cf1StoreFileCount = -1;
    int cf2StoreFileCount = -1;
    int cf3StoreFileCount = -1;
    int cf1StoreFileCount1 = -1;
    int cf2StoreFileCount1 = -1;
    int cf3StoreFileCount1 = -1;
    try {
      TEST_UTIL.startMiniCluster(1);
      TEST_UTIL.getAdmin()
        .createNamespace(NamespaceDescriptor.create(TABLENAME.getNamespaceAsString()).build());
      TEST_UTIL.getAdmin().createTable(tableDescriptor);
      TEST_UTIL.waitTableAvailable(TABLENAME);
      Connection conn = ConnectionFactory.createConnection(conf);
      Table table = conn.getTable(TABLENAME);
      doPut(table, memstoreFlushSize);
      table.close();
      conn.close();

      Region region = getRegionWithName(TABLENAME).getFirst();
      cf1StoreFileCount = region.getStore(FAMILY1).getStorefilesCount();
      cf2StoreFileCount = region.getStore(FAMILY2).getStorefilesCount();
      cf3StoreFileCount = region.getStore(FAMILY3).getStorefilesCount();
    } finally {
      TEST_UTIL.shutdownMiniCluster();
    }

    LOG.info("==============Test with selective flush enabled===============");
    conf.set(FlushPolicyFactory.HBASE_FLUSH_POLICY_KEY, FlushAllLargeStoresPolicy.class.getName());
    // default value of per-cf flush lower bound is too big, set to a small enough value
    conf.setLong(FlushLargeStoresPolicy.HREGION_COLUMNFAMILY_FLUSH_SIZE_LOWER_BOUND_MIN, 0);
    try {
      TEST_UTIL.startMiniCluster(1);
      TEST_UTIL.getAdmin()
        .createNamespace(NamespaceDescriptor.create(TABLENAME.getNamespaceAsString()).build());
      TEST_UTIL.getAdmin().createTable(tableDescriptor);
      Connection conn = ConnectionFactory.createConnection(conf);
      Table table = conn.getTable(TABLENAME);
      doPut(table, memstoreFlushSize);
      table.close();
      conn.close();

      Region region = getRegionWithName(TABLENAME).getFirst();
      cf1StoreFileCount1 = region.getStore(FAMILY1).getStorefilesCount();
      cf2StoreFileCount1 = region.getStore(FAMILY2).getStorefilesCount();
      cf3StoreFileCount1 = region.getStore(FAMILY3).getStorefilesCount();
    } finally {
      TEST_UTIL.shutdownMiniCluster();
    }

    LOG.info("disable selective flush: " + Bytes.toString(FAMILY1) + "=>" + cf1StoreFileCount + ", "
      + Bytes.toString(FAMILY2) + "=>" + cf2StoreFileCount + ", " + Bytes.toString(FAMILY3) + "=>"
      + cf3StoreFileCount);
    LOG.info("enable selective flush: " + Bytes.toString(FAMILY1) + "=>" + cf1StoreFileCount1 + ", "
      + Bytes.toString(FAMILY2) + "=>" + cf2StoreFileCount1 + ", " + Bytes.toString(FAMILY3) + "=>"
      + cf3StoreFileCount1);
    // small CF will have less store files.
    assertTrue(cf1StoreFileCount1 < cf1StoreFileCount);
    assertTrue(cf2StoreFileCount1 < cf2StoreFileCount);
  }
When a log roll is about to happen, we do a flush of the regions who will be affected by the
log roll. These flushes cannot be a selective flushes, otherwise we cannot roll the logs. This
test ensures that we do a full-flush in that scenario.
hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestPerColumnFamilyFlush.java