HBASE-30398 Support configuring the concurrency of off-peak compactions

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

@@ -43,7 +43,6 @@
 import java.util.concurrent.ExecutorCompletionService;
 import java.util.concurrent.Future;
 import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.LongAdder;
@@ -88,6 +87,7 @@
 import org.apache.hadoop.hbase.regionserver.compactions.CompactionLifeCycleTracker;
 import org.apache.hadoop.hbase.regionserver.compactions.CompactionProgress;
 import org.apache.hadoop.hbase.regionserver.compactions.CompactionRequestImpl;
+import org.apache.hadoop.hbase.regionserver.compactions.OffPeakCompactionTracker;
 import org.apache.hadoop.hbase.regionserver.compactions.OffPeakHours;
 import org.apache.hadoop.hbase.regionserver.querymatcher.ScanQueryMatcher;
 import org.apache.hadoop.hbase.regionserver.storefiletracker.StoreFileTracker;
@@ -199,7 +199,8 @@ public class HStore
 
   final StoreEngine<?, ?, ?, ?> storeEngine;
 
-  private static final AtomicBoolean offPeakCompactionTracker = new AtomicBoolean();
+  private static final OffPeakCompactionTracker offPeakCompactionTracker =
+    OffPeakCompactionTracker.getInstance();
   private volatile OffPeakHours offPeakHours;
 
   private static final int DEFAULT_FLUSH_RETRIES_NUMBER = 10;
@@ -1500,20 +1501,20 @@ public Optional<CompactionContext> requestCompaction(int priority,
         if (!compaction.hasSelection()) {
           boolean isUserCompaction = priority == Store.PRIORITY_USER;
           boolean mayUseOffPeak =
-            offPeakHours.isOffPeakHour() && offPeakCompactionTracker.compareAndSet(false, true);
+            offPeakHours.isOffPeakHour() && offPeakCompactionTracker.tryAcquire();
           try {
             compaction.select(this.filesCompacting, isUserCompaction, mayUseOffPeak,
               forceMajor && filesCompacting.isEmpty());
           } catch (IOException e) {
             if (mayUseOffPeak) {
-              offPeakCompactionTracker.set(false);
+              offPeakCompactionTracker.release();
             }
             throw e;
           }
           assert compaction.hasSelection();
           if (mayUseOffPeak && !compaction.getRequest().isOffPeak()) {
             // Compaction policy doesn't want to take advantage of off-peak.
-            offPeakCompactionTracker.set(false);
+            offPeakCompactionTracker.release();
           }
         }
         if (this.getCoprocessorHost() != null) {
@@ -1623,7 +1624,7 @@ public void cancelRequestedCompaction(CompactionContext compaction) {
   private void finishCompactionRequest(CompactionRequestImpl cr) {
     this.region.reportCompactionRequestEnd(cr.isMajor(), cr.getFiles().size(), cr.getSize());
     if (cr.isOffPeak()) {
-      offPeakCompactionTracker.set(false);
+      offPeakCompactionTracker.release();
       cr.setOffPeak(false);
     }
     synchronized (filesCompacting) {
@@ -2220,6 +2221,7 @@ public void registerChildren(ConfigurationManager manager) {
     if (cacheConfig != null) {
       manager.registerObserver(cacheConfig);
     }
+    offPeakCompactionTracker.registerIfNeeded(manager, conf);
   }
 
   /**

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

@@ -65,6 +65,9 @@ public class CompactionConfiguration {
     ConfigKey.LONG("hbase.hstore.compaction.max.size");
   public static final String HBASE_HSTORE_COMPACTION_MAX_SIZE_OFFPEAK_KEY =
     ConfigKey.LONG("hbase.hstore.compaction.max.size.offpeak");
+  public static final String HBASE_HSTORE_OFFPEAK_COMPACTION_CONCURRENCY_KEY =
+    ConfigKey.INT("hbase.hstore.offpeak.compaction.concurrency");
+  public static final int DEFAULT_OFFPEAK_COMPACTION_CONCURRENCY = 1;
   public static final String HBASE_HSTORE_OFFPEAK_END_HOUR =
     ConfigKey.INT("hbase.offpeak.end.hour");
   public static final String HBASE_HSTORE_OFFPEAK_START_HOUR =

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

@@ -0,0 +1,92 @@
+/**
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hbase.regionserver.compactions;
+
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.conf.ConfigurationManager;
+import org.apache.hadoop.hbase.conf.ConfigurationObserver;
+import org.apache.yetus.audience.InterfaceAudience;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@InterfaceAudience.Private
+public class OffPeakCompactionTracker extends Semaphore implements ConfigurationObserver {
+  private static final Logger LOG = LoggerFactory.getLogger(OffPeakCompactionTracker.class);
+
+  private static final OffPeakCompactionTracker INSTANCE = new OffPeakCompactionTracker(
+    CompactionConfiguration.DEFAULT_OFFPEAK_COMPACTION_CONCURRENCY);
+
+  private final AtomicBoolean registered;
+
+  private int concurrency;
+
+  public OffPeakCompactionTracker(int concurrency) {
+    super(concurrency);
+    this.concurrency = concurrency;
+    this.registered = new AtomicBoolean(false);
+  }
+
+  public static OffPeakCompactionTracker getInstance() {
+    return INSTANCE;
+  }
+
+  public void registerIfNeeded(ConfigurationManager manager, Configuration conf) {
+    if (registered.get()) {
+      return;
+    }
+    if (registered.compareAndSet(false, true)) {
+      manager.registerObserver(this);
+      updateConcurrency(conf);
+    }
+  }
+
+  public void updateConcurrency(Configuration conf) {
+    int newVal = conf.getInt(
+                 CompactionConfiguration.HBASE_HSTORE_OFFPEAK_COMPACTION_CONCURRENCY_KEY,
+                 CompactionConfiguration.DEFAULT_OFFPEAK_COMPACTION_CONCURRENCY);
+
+    if (this.concurrency == newVal) {
+      return;
+    }
+
+    LOG.info("Changing the value of {} from {} to {}",
+              CompactionConfiguration.HBASE_HSTORE_OFFPEAK_COMPACTION_CONCURRENCY_KEY,
+              this.concurrency,
+              newVal);
+
+    int delta = newVal - this.concurrency;
+    if (delta > 0) {
+      release(delta);
+    }
+
+    if (delta < 0) {
+      reducePermits(-delta);
+    }
+
+    this.concurrency = newVal;
+  }
+
+  @Override
+  public void onConfigurationChange(Configuration conf) {
+    updateConcurrency(conf);
+  }
+}

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

@@ -0,0 +1,130 @@
+/**
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hbase.regionserver.compactions;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.conf.ConfigurationManager;
+import org.apache.hadoop.hbase.regionserver.HStore;
+import org.apache.hadoop.hbase.testclassification.RegionServerTests;
+import org.apache.hadoop.hbase.testclassification.SmallTests;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+@Tag(RegionServerTests.TAG)
+@Tag(SmallTests.TAG)
+public class TestOffPeakCompactionTracker {
+
+  @Test
+  public void testOffPeakCompactionTracker() throws Exception {
+    OffPeakCompactionTracker offPeakCompactionTracker = OffPeakCompactionTracker.getInstance();
+    Configuration conf = new Configuration();
+    conf.setInt(CompactionConfiguration.HBASE_HSTORE_OFFPEAK_COMPACTION_CONCURRENCY_KEY, 5);
+    offPeakCompactionTracker.updateConcurrency(conf);
+    OffPeakHours offPeakHours = Mockito.mock(OffPeakHours.class);
+    Mockito.when(offPeakHours.isOffPeakHour()).thenReturn(true);
+
+    AtomicInteger count = new AtomicInteger(0);
+    int familyNum = 10;
+    List<HStore> stores = new ArrayList<>();
+    for (int i = 0; i < familyNum; i++) {
+      HStore store = Mockito.mock(HStore.class);
+      Mockito.doAnswer(invocation -> {
+        boolean mayUseOffPeak = offPeakHours.isOffPeakHour() &&
+          offPeakCompactionTracker.tryAcquire();
+        if (mayUseOffPeak) {
+          count.incrementAndGet();
+        }
+        return null;
+      }).when(store).requestCompaction();
+      stores.add(store);
+    }
+
+    CountDownLatch startLatch = new CountDownLatch(1);
+    CountDownLatch doneLatch = new CountDownLatch(familyNum);
+
+    for (int i = 0; i < familyNum; i++) {
+      HStore store = stores.get(i);
+      CompletableFuture.supplyAsync(() -> {
+        try {
+          startLatch.await();
+          store.requestCompaction();
+          doneLatch.countDown();
+        } catch (Exception e) {
+          throw new RuntimeException(e);
+        }
+        return null;
+      });
+    }
+
+    assertEquals(5, offPeakCompactionTracker.availablePermits());
+
+    startLatch.countDown();
+    doneLatch.await();
+    assertEquals(5, count.get());
+    assertEquals(0, offPeakCompactionTracker.availablePermits());
+    offPeakCompactionTracker.release(count.get());
+  }
+
+  @Test
+  public void testOffPeakCompactionConcurrencyOnChange() throws Exception {
+    OffPeakCompactionTracker offPeakCompactionTracker = OffPeakCompactionTracker.getInstance();
+    OffPeakCompactionTracker spyTracker = Mockito.spy(offPeakCompactionTracker);
+    ConfigurationManager manager = new ConfigurationManager();
+    Configuration conf = new Configuration();
+    conf.setInt(CompactionConfiguration.HBASE_HSTORE_OFFPEAK_COMPACTION_CONCURRENCY_KEY, 2);
+
+    int familyNum = 10;
+    List<HStore> stores = new ArrayList<>();
+    for (int i = 0; i < familyNum; i++) {
+      HStore store = Mockito.mock(HStore.class);
+      Mockito.doAnswer(invocation -> {
+        spyTracker.registerIfNeeded(manager, conf);
+        return null;
+      }).when(store).registerChildren(manager);
+      stores.add(store);
+    }
+    for (HStore store : stores) {
+      store.registerChildren(manager);
+    }
+
+    Mockito.verify(spyTracker, Mockito.times(1))
+           .updateConcurrency(Mockito.any(Configuration.class));
+    assertEquals(2, spyTracker.availablePermits());
+
+    conf.setInt(CompactionConfiguration.HBASE_HSTORE_OFFPEAK_COMPACTION_CONCURRENCY_KEY, 100);
+    manager.notifyAllObservers(conf);
+    Mockito.verify(spyTracker, Mockito.times(2))
+           .updateConcurrency(Mockito.any(Configuration.class));
+    assertEquals(100, spyTracker.availablePermits());
+
+    conf.setInt(CompactionConfiguration.HBASE_HSTORE_OFFPEAK_COMPACTION_CONCURRENCY_KEY, 10);
+    manager.notifyAllObservers(conf);
+    Mockito.verify(spyTracker, Mockito.times(3))
+           .updateConcurrency(Mockito.any(Configuration.class));
+    assertEquals(10, spyTracker.availablePermits());
+  }
+}