HBASE-27944:HStore.needsCompaction() should return false if we disable comapction against a table

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HStore.java

@@ -2065,6 +2065,9 @@ public void abort() throws IOException {
 
   @Override
   public boolean needsCompaction() {
+    if (!region.getTableDescriptor().isCompactionEnabled()) {
+      return false;
+    }
     List<HStoreFile> filesCompactingClone = null;
     synchronized (filesCompacting) {
       filesCompactingClone = Lists.newArrayList(filesCompacting);

hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHStore.java

@@ -983,6 +983,63 @@ public void testStoreUsesConfigurationFromHcdAndHtd() throws Exception {
     assertFalse(store.throttleCompaction(anyValue));
   }
 
+  @Test
+  public void testNeedsCompaction() throws IOException {
+    byte[] value = new byte[9];
+    final long timestamp = EnvironmentEdgeManager.currentTime();
+    final long seqId = 100;
+    final Cell cell = createCell(qf2, timestamp, seqId, value);
+
+    Configuration conf = HBaseConfiguration.create();
+    conf.set("hbase.hstore.compactionThreshold", "3");
+    conf.setBoolean(WALFactory.WAL_ENABLED, false);
+    int minFilesToCompact = conf.getInt("hbase.hstore.compactionThreshold", 3);
+
+    // table which enables compaction
+    // HStore.needsCompaction would return false if the number of store files less than
+    // hbase.hstore.compactionThreshold
+    // HStore.needsCompaction would return true if the number of store files greater than or quals
+    // to hbase.hstore.compactionThreshold
+    TableDescriptorBuilder builder =
+      TableDescriptorBuilder.newBuilder(TableName.valueOf(table)).setSplitEnabled(false);
+    init(name.getMethodName(), conf, builder,
+      ColumnFamilyDescriptorBuilder.newBuilder(family).build());
+    LOG.info("The table has already enabled Compaction.");
+    MemStoreSizing memStoreSizing = new NonThreadSafeMemStoreSizing();
+    store.add(cell, memStoreSizing);
+    flushStore(store, id++);
+    store.add(cell, memStoreSizing);
+    flushStore(store, id++);
+    store.add(cell, memStoreSizing);
+    assertTrue(store.getStorefiles().size() < minFilesToCompact);
+    assertEquals(false, store.needsCompaction());
+
+    store.add(cell, memStoreSizing);
+    flushStore(store, id++);
+    assertTrue(store.getStorefiles().size() >= minFilesToCompact);
+    assertEquals(true, store.needsCompaction());
+
+    // table which disable compaction
+    // HStore.needsCompaction would return false no matter how much store files there are
+    TableDescriptorBuilder builderDisableCompaction = TableDescriptorBuilder
+      .newBuilder(TableName.valueOf(table)).setSplitEnabled(false).setCompactionEnabled(false);
+    init(name.getMethodName(), conf, builderDisableCompaction,
+      ColumnFamilyDescriptorBuilder.newBuilder(family).build());
+    LOG.info("The table has already disabled Compaction.");
+    MemStoreSizing memStoreSizingDisableCompaction = new NonThreadSafeMemStoreSizing();
+    store.add(cell, memStoreSizingDisableCompaction);
+    flushStore(store, id++);
+    store.add(cell, memStoreSizingDisableCompaction);
+    flushStore(store, id++);
+    assertTrue(store.getStorefiles().size() < minFilesToCompact);
+    assertEquals(false, store.needsCompaction());
+
+    store.add(cell, memStoreSizingDisableCompaction);
+    flushStore(store, id++);
+    assertTrue(store.getStorefiles().size() >= minFilesToCompact);
+    assertEquals(false, store.needsCompaction());
+  }
+
   public static class DummyStoreEngine extends DefaultStoreEngine {
     public static DefaultCompactor lastCreatedCompactor = null;