diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HStore.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HStore.java index 25518a89f490..7c535dcc8675 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HStore.java +++ b/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 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); } /** diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/compactions/CompactionConfiguration.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/compactions/CompactionConfiguration.java index 59fafc9908ab..ae470bab54b9 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/compactions/CompactionConfiguration.java +++ b/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 = diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/compactions/OffPeakCompactionTracker.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/compactions/OffPeakCompactionTracker.java new file mode 100644 index 000000000000..1e30994623e5 --- /dev/null +++ b/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); + } +} diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/compactions/TestOffPeakCompactionTracker.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/compactions/TestOffPeakCompactionTracker.java new file mode 100644 index 000000000000..8a9f27076968 --- /dev/null +++ b/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 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 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()); + } +}