HBASE-8272 make compaction checker frequency configurable per table/cf

git-svn-id: https://svn.apache.org/repos/asf/hbase/trunk@1477425 13f79535-47bb-0310-9956-ffa450edef68

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

@@ -682,12 +682,9 @@ public class HRegionServer implements ClientProtocol,
     // Compaction thread
     this.compactSplitThread = new CompactSplitThread(this);
 
-    // Background thread to check for compactions; needed if region
-    // has not gotten updates in a while. Make it run at a lesser frequency.
-    int multiplier = this.conf.getInt(HConstants.THREAD_WAKE_FREQUENCY +
-      ".multiplier", 1000);
-    this.compactionChecker = new CompactionChecker(this,
-      this.threadWakeFrequency * multiplier, this);
+    // Background thread to check for compactions; needed if region has not gotten updates
+    // in a while. It will take care of not checking too frequently on store-by-store basis.
+    this.compactionChecker = new CompactionChecker(this, this.threadWakeFrequency, this);
     this.periodicFlusher = new PeriodicMemstoreFlusher(this.threadWakeFrequency, this);
     // Health checker thread.
     int sleepTime = this.conf.getInt(HConstants.HEALTH_CHORE_WAKE_FREQ,
@@ -1236,6 +1233,7 @@ public class HRegionServer implements ClientProtocol,
     private final HRegionServer instance;
     private final int majorCompactPriority;
     private final static int DEFAULT_PRIORITY = Integer.MAX_VALUE;
+    private long iteration = 0;
 
     CompactionChecker(final HRegionServer h, final int sleepTime,
         final Stoppable stopper) {
@@ -1258,6 +1256,9 @@ public class HRegionServer implements ClientProtocol,
           continue;
         for (Store s : r.getStores().values()) {
           try {
+            long multiplier = s.getCompactionCheckMultiplier();
+            assert multiplier > 0;
+            if (iteration % multiplier != 0) continue;
             if (s.needsCompaction()) {
               // Queue a compaction. Will recognize if major is needed.
               this.instance.compactSplitThread.requestCompaction(r, s, getName()
@@ -1278,6 +1279,7 @@ public class HRegionServer implements ClientProtocol,
           }
         }
       }
+      iteration = (iteration == Long.MAX_VALUE) ? 0 : (iteration + 1);
     }
   }
 

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

@@ -106,7 +106,10 @@ import com.google.common.collect.Lists;
  */
 @InterfaceAudience.Private
 public class HStore implements Store {
+  public static final String COMPACTCHECKER_INTERVAL_MULTIPLIER_KEY =
+      "hbase.server.compactchecker.interval.multiplier";
   public static final String BLOCKING_STOREFILES_KEY = "hbase.hstore.blockingStoreFiles";
+  public static final int DEFAULT_COMPACTCHECKER_INTERVAL_MULTIPLIER = 1000;
   public static final int DEFAULT_BLOCKING_STOREFILE_COUNT = 7;
 
   static final Log LOG = LogFactory.getLog(HStore.class);
@@ -165,6 +168,7 @@ public class HStore implements Store {
   private int pauseTime;
 
   private long blockingFileCount;
+  private int compactionCheckMultiplier;
 
   /**
    * Constructor
@@ -219,6 +223,13 @@ public class HStore implements Store {
 
     this.blockingFileCount =
         conf.getInt(BLOCKING_STOREFILES_KEY, DEFAULT_BLOCKING_STOREFILE_COUNT);
+    this.compactionCheckMultiplier = conf.getInt(
+        COMPACTCHECKER_INTERVAL_MULTIPLIER_KEY, DEFAULT_COMPACTCHECKER_INTERVAL_MULTIPLIER);
+    if (this.compactionCheckMultiplier <= 0) {
+      LOG.error("Compaction check period multiplier must be positive, setting default: "
+          + DEFAULT_COMPACTCHECKER_INTERVAL_MULTIPLIER);
+      this.compactionCheckMultiplier = DEFAULT_COMPACTCHECKER_INTERVAL_MULTIPLIER;
+    }
 
     if (HStore.closeCheckInterval == 0) {
       HStore.closeCheckInterval = conf.getInt(
@@ -291,6 +302,11 @@ public class HStore implements Store {
     return this.region.memstoreFlushSize;
   }
 
+  @Override
+  public long getCompactionCheckMultiplier() {
+    return this.compactionCheckMultiplier;
+  }
+
   public long getBlockingFileCount() {
     return blockingFileCount;
   }
@@ -1829,7 +1845,7 @@ public class HStore implements Store {
 
   public static final long FIXED_OVERHEAD =
       ClassSize.align(ClassSize.OBJECT + (15 * ClassSize.REFERENCE) + (4 * Bytes.SIZEOF_LONG)
-              + (4 * Bytes.SIZEOF_INT) + (2 * Bytes.SIZEOF_BOOLEAN));
+              + (5 * Bytes.SIZEOF_INT) + (2 * Bytes.SIZEOF_BOOLEAN));
 
   public static final long DEEP_OVERHEAD = ClassSize.align(FIXED_OVERHEAD
       + ClassSize.OBJECT + ClassSize.REENTRANT_LOCK

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

@@ -31,16 +31,23 @@ import org.apache.hadoop.classification.InterfaceStability;
 @InterfaceStability.Unstable
 public interface StoreConfigInformation {
   /**
-   * Gets the Memstore flush size for the region that this store works with.
-   * TODO: remove after HBASE-7236 is fixed.
+   * TODO: remove after HBASE-7252 is fixed.
+   * @return Gets the Memstore flush size for the region that this store works with.
    */
   public long getMemstoreFlushSize();
 
   /**
-   * Gets the cf-specific time-to-live for store files.
+   * @return Gets the cf-specific time-to-live for store files.
    */
   public long getStoreFileTtl();
 
+  /**
+   * @return Gets the cf-specific compaction check frequency multiplier.
+   *         The need for compaction (outside of normal checks during flush, open, etc.) will
+   *         be ascertained every multiplier * HConstants.THREAD_WAKE_FREQUENCY milliseconds.
+   */
+  public long getCompactionCheckMultiplier();
+
   /**
    * The number of files required before flushes for this store will be blocked.
    */