@Test
  public void testFlushingWhenLogRolling() throws Exception {
    TableName tableName = TableName.valueOf("testFlushingWhenLogRolling");
    Configuration conf = TEST_UTIL.getConfiguration();
    conf.setLong(HConstants.HREGION_MEMSTORE_FLUSH_SIZE, 128 * 1024 * 1024);
    conf.set(FlushPolicyFactory.HBASE_FLUSH_POLICY_KEY, FlushAllLargeStoresPolicy.class.getName());
    long cfFlushSizeLowerBound = 2048;
    conf.setLong(FlushLargeStoresPolicy.HREGION_COLUMNFAMILY_FLUSH_SIZE_LOWER_BOUND_MIN,
      cfFlushSizeLowerBound);

    // One hour, prevent periodic rolling
    conf.setLong("hbase.regionserver.logroll.period", 60L * 60 * 1000);
    // prevent rolling by size
    conf.setLong("hbase.regionserver.hlog.blocksize", 128L * 1024 * 1024);
    // Make it 10 as max logs before a flush comes on.
    final int maxLogs = 10;
    conf.setInt("hbase.regionserver.maxlogs", maxLogs);

    final int numRegionServers = 1;
    TEST_UTIL.startMiniCluster(numRegionServers);
    try {
      Table table = TEST_UTIL.createTable(tableName, FAMILIES);
      Pair<HRegion, HRegionServer> desiredRegionAndServer = getRegionWithName(tableName);
      final HRegion desiredRegion = desiredRegionAndServer.getFirst();
      assertTrue("Could not find a region which hosts the new region.", desiredRegion != null);
      LOG.info("Writing to region=" + desiredRegion);

      // Add one row for both CFs.
      for (int i = 1; i <= 3; i++) {
        table.put(createPut(i, 0));
      }
      // Now only add row to CF1, make sure when we force a flush, CF1 is larger than the lower
      // bound and CF2 and CF3 are smaller than the lower bound.
      for (int i = 0; i < maxLogs; i++) {
        for (int j = 0; j < 100; j++) {
          table.put(createPut(1, i * 100 + j));
        }
        // Roll the WAL. The log file count is less than maxLogs so no flush is triggered.
        int currentNumRolledLogFiles = getNumRolledLogFiles(desiredRegion);
        assertNull(getWAL(desiredRegion).rollWriter());
        TEST_UTIL.waitFor(60000,
          () -> getNumRolledLogFiles(desiredRegion) > currentNumRolledLogFiles);
      }
      assertEquals(maxLogs, getNumRolledLogFiles(desiredRegion));
      assertTrue(
        desiredRegion.getStore(FAMILY1).getMemStoreSize().getHeapSize() > cfFlushSizeLowerBound);
      assertTrue(
        desiredRegion.getStore(FAMILY2).getMemStoreSize().getHeapSize() < cfFlushSizeLowerBound);
      assertTrue(
        desiredRegion.getStore(FAMILY3).getMemStoreSize().getHeapSize() < cfFlushSizeLowerBound);
      table.put(createPut(1, 12345678));
      // Make numRolledLogFiles greater than maxLogs
      desiredRegionAndServer.getSecond().getWalRoller().requestRollAll();
      // Wait for some time till the flush caused by log rolling happens.
      TEST_UTIL.waitFor(30000, new Waiter.ExplainingPredicate<Exception>() {

        @Override
        public boolean evaluate() throws Exception {
          return desiredRegion.getMemStoreDataSize() == 0;
        }

        @Override
        public String explainFailure() throws Exception {
          long memstoreSize = desiredRegion.getMemStoreDataSize();
          if (memstoreSize > 0) {
            return "Still have unflushed entries in memstore, memstore size is " + memstoreSize;
          }
          return "Unknown";
        }
      });
      LOG.info("Finished waiting on flush after too many WALs...");
      // Individual families should have been flushed.
      assertEquals(MutableSegment.DEEP_OVERHEAD,
        desiredRegion.getStore(FAMILY1).getMemStoreSize().getHeapSize());
      assertEquals(MutableSegment.DEEP_OVERHEAD,
        desiredRegion.getStore(FAMILY2).getMemStoreSize().getHeapSize());
      assertEquals(MutableSegment.DEEP_OVERHEAD,
        desiredRegion.getStore(FAMILY3).getMemStoreSize().getHeapSize());
      // let WAL cleanOldLogs
      assertNull(getWAL(desiredRegion).rollWriter(true));
      TEST_UTIL.waitFor(60000, () -> getNumRolledLogFiles(desiredRegion) < maxLogs);
    } finally {
      TEST_UTIL.shutdownMiniCluster();
    }
  }
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