diff --git hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreFlusher.java hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreFlusher.java index a69d8c0..843548d 100644 --- hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreFlusher.java +++ hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreFlusher.java @@ -438,7 +438,11 @@ class MemStoreFlusher implements FlushRequester { Region region = fqe.region; if (!region.getRegionInfo().isMetaRegion() && isTooManyStoreFiles(region)) { - if (fqe.isMaximumWait(this.blockingWaitTime)) { + // If compaction is disabled on a table, do not wait + if (!region.getTableDesc().isCompactionEnabled()) { + LOG.info("Compaction is disabled, proceed with flush of " + + region.getRegionInfo().getRegionNameAsString()); + } else if (fqe.isMaximumWait(this.blockingWaitTime)) { LOG.info("Waited " + (EnvironmentEdgeManager.currentTime() - fqe.createTime) + "ms on a compaction to clean up 'too many store files'; waited " + "long enough... proceeding with flush of " + diff --git hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestCompactSplitThread.java hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestCompactSplitThread.java index 3457883..4a3540a 100644 --- hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestCompactSplitThread.java +++ hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestCompactSplitThread.java @@ -24,8 +24,12 @@ import org.apache.hadoop.hbase.*; import org.apache.hadoop.hbase.client.*; import org.apache.hadoop.hbase.testclassification.MediumTests; import org.apache.hadoop.hbase.util.Bytes; +import org.junit.After; +import org.junit.AfterClass; import org.junit.Assert; +import org.junit.BeforeClass; import org.junit.Test; + import org.junit.experimental.categories.Category; import static org.junit.Assert.assertEquals; @@ -36,15 +40,55 @@ public class TestCompactSplitThread { private static final HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility(); private final TableName tableName = TableName.valueOf(getClass().getSimpleName()); private final byte[] family = Bytes.toBytes("f"); + private static final int NUM_RS = 1; + private static final int blockingStoreFiles = 3; - @Test - public void testThreadPoolSizeTuning() throws Exception { - Configuration conf = TEST_UTIL.getConfiguration(); + + /** + * Setup the config for the cluster + */ + @BeforeClass + public static void setupCluster() throws Exception { + setupConf(TEST_UTIL.getConfiguration()); + TEST_UTIL.startMiniCluster(NUM_RS); + } + + private static void setupConf(Configuration conf) { + // disable the ui + conf.setInt("hbase.regionsever.info.port", -1); + // change the flush size to a small amount, regulating number of store files + conf.setInt("hbase.hregion.memstore.flush.size", 25000); + // so make sure we get a compaction when doing a load, but keep around some + // files in the store + conf.setInt("hbase.hstore.compaction.min", 2); + conf.setInt("hbase.hstore.compactionThreshold", 5); + // block writes if we get to 12 store files + conf.setInt("hbase.hstore.blockingStoreFiles", blockingStoreFiles); + // Ensure no extra cleaners on by default (e.g. TimeToLiveHFileCleaner) conf.setInt(CompactSplitThread.LARGE_COMPACTION_THREADS, 3); conf.setInt(CompactSplitThread.SMALL_COMPACTION_THREADS, 4); conf.setInt(CompactSplitThread.SPLIT_THREADS, 5); conf.setInt(CompactSplitThread.MERGE_THREADS, 6); - TEST_UTIL.startMiniCluster(1); + } + + @After + public void tearDown() throws Exception { + TEST_UTIL.deleteTable(tableName); + } + + @AfterClass + public static void cleanupTest() throws Exception { + try { + TEST_UTIL.shutdownMiniCluster(); + } catch (Exception e) { + // NOOP; + } + } + + @Test + public void testThreadPoolSizeTuning() throws Exception { + Configuration conf = TEST_UTIL.getConfiguration(); + //TEST_UTIL.startMiniCluster(1); Connection conn = ConnectionFactory.createConnection(conf); try { HTableDescriptor htd = new HTableDescriptor(tableName); @@ -95,7 +139,21 @@ public class TestCompactSplitThread { assertEquals(5, regionServer.compactSplitThread.getMergeThreadNum()); } finally { conn.close(); - TEST_UTIL.shutdownMiniCluster(); + } + } + + @Test(timeout = 60000) + public void testFlushWithTableCompactionDisabled() throws Exception { + Admin admin = TEST_UTIL.getHBaseAdmin(); + + HTableDescriptor htd = new HTableDescriptor(tableName); + htd.setCompactionEnabled(false); + TEST_UTIL.createTable(htd, new byte[][] { family }, null); + + // load the table + for (int i = 0; i < blockingStoreFiles + 1; i ++) { + TEST_UTIL.loadTable(TEST_UTIL.getConnection().getTable(tableName), family); + TEST_UTIL.flush(tableName); } } }