HBASE-14387 Compaction improvements: Maximum off-peak compaction size (Vladimir Rodionov)

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

@@ -57,6 +57,8 @@ public class CompactionConfiguration {
   public static final String HBASE_HSTORE_COMPACTION_MAX_KEY = "hbase.hstore.compaction.max";
   public static final String HBASE_HSTORE_COMPACTION_MAX_SIZE_KEY =
     "hbase.hstore.compaction.max.size";
+  public static final String HBASE_HSTORE_COMPACTION_MAX_SIZE_OFFPEAK_KEY =
+      "hbase.hstore.compaction.max.size.offpeak";
   public static final String HBASE_HSTORE_OFFPEAK_END_HOUR = "hbase.offpeak.end.hour";
   public static final String HBASE_HSTORE_OFFPEAK_START_HOUR = "hbase.offpeak.start.hour";
   public static final String HBASE_HSTORE_MIN_LOCALITY_TO_SKIP_MAJOR_COMPACT =
@@ -68,6 +70,7 @@ public class CompactionConfiguration {
   private final double offPeakCompactionRatio;
   /** Since all these properties can change online, they are volatile **/
   private final long maxCompactSize;
+  private final long offPeakMaxCompactSize;
   private final long minCompactSize;
   private final int minFilesToCompact;
   private final int maxFilesToCompact;
@@ -82,6 +85,8 @@ public class CompactionConfiguration {
     this.storeConfigInfo = storeConfigInfo;
 
     maxCompactSize = conf.getLong(HBASE_HSTORE_COMPACTION_MAX_SIZE_KEY, Long.MAX_VALUE);
+    offPeakMaxCompactSize = conf.getLong(HBASE_HSTORE_COMPACTION_MAX_SIZE_OFFPEAK_KEY, 
+      maxCompactSize);      
     minCompactSize = conf.getLong(HBASE_HSTORE_COMPACTION_MIN_SIZE_KEY,
         storeConfigInfo.getMemstoreFlushSize());
     minFilesToCompact = Math.max(2, conf.getInt(HBASE_HSTORE_COMPACTION_MIN_KEY,
@@ -102,10 +107,11 @@ public class CompactionConfiguration {
   @Override
   public String toString() {
     return String.format(
-      "size [%d, %d); files [%d, %d); ratio %f; off-peak ratio %f; throttle point %d;"
+      "size [%d, %d, %d); files [%d, %d); ratio %f; off-peak ratio %f; throttle point %d;"
       + " major period %d, major jitter %f, min locality to compact %f",
       minCompactSize,
       maxCompactSize,
+      offPeakMaxCompactSize,
       minFilesToCompact,
       maxFilesToCompact,
       compactionRatio,
@@ -189,4 +195,16 @@ public class CompactionConfiguration {
   public float getMinLocalityToForceCompact() {
     return minLocalityToForceCompact;
   }
+
+  public long getOffPeakMaxCompactSize() {
+    return offPeakMaxCompactSize;
+  }
+  
+  public long getMaxCompactSize(boolean mayUseOffpeak) {
+    if (mayUseOffpeak) {
+      return getOffPeakMaxCompactSize();
+    } else {
+      return getMaxCompactSize();
+    }
+  }
 }

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

@@ -96,7 +96,7 @@ public class ExploringCompactionPolicy extends RatioBasedCompactionPolicy {
           smallestSize = size;
         }
 
-        if (size > comConf.getMaxCompactSize()) {
+        if (size > comConf.getMaxCompactSize(mayUseOffPeak)) {
           continue;
         }
 

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

@@ -96,7 +96,7 @@ public class RatioBasedCompactionPolicy extends CompactionPolicy {
     // If we can't have all files, we cannot do major anyway
     boolean isAllFiles = candidateFiles.size() == candidateSelection.size();
     if (!(forceMajor && isAllFiles)) {
-      candidateSelection = skipLargeFiles(candidateSelection);
+      candidateSelection = skipLargeFiles(candidateSelection, mayUseOffPeak);
       isAllFiles = candidateFiles.size() == candidateSelection.size();
     }
 
@@ -128,10 +128,11 @@ public class RatioBasedCompactionPolicy extends CompactionPolicy {
    * exclude all files above maxCompactSize
    * Also save all references. We MUST compact them
    */
-  private ArrayList<StoreFile> skipLargeFiles(ArrayList<StoreFile> candidates) {
+  private ArrayList<StoreFile> skipLargeFiles(ArrayList<StoreFile> candidates, 
+    boolean mayUseOffpeak) {
     int pos = 0;
     while (pos < candidates.size() && !candidates.get(pos).isReference()
-      && (candidates.get(pos).getReader().length() > comConf.getMaxCompactSize())) {
+      && (candidates.get(pos).getReader().length() > comConf.getMaxCompactSize(mayUseOffpeak))) {
       ++pos;
     }
     if (pos > 0) {

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

@@ -194,6 +194,14 @@ public class TestRegionServerOnlineConfigChange {
     rs1.getConfigurationManager().notifyAllObservers(conf);
     assertEquals(newMaxCompactSize,
                  hstore.getStoreEngine().getCompactionPolicy().getConf().getMaxCompactSize());
+    // Check if the offPeakMaxCompactSize gets updated.
+    long newOffpeakMaxCompactSize =
+            hstore.getStoreEngine().getCompactionPolicy().getConf().getOffPeakMaxCompactSize() - 1;
+    conf.setLong(CompactionConfiguration.HBASE_HSTORE_COMPACTION_MAX_SIZE_OFFPEAK_KEY,
+      newOffpeakMaxCompactSize);
+    rs1.getConfigurationManager().notifyAllObservers(conf);
+    assertEquals(newOffpeakMaxCompactSize,
+                 hstore.getStoreEngine().getCompactionPolicy().getConf().getOffPeakMaxCompactSize());
 
     // Check if majorCompactionPeriod gets updated.
     long newMajorCompactionPeriod =