HBASE-19528 - Major Compaction Tool
Project: http://git-wip-us.apache.org/repos/asf/hbase/repo Commit: http://git-wip-us.apache.org/repos/asf/hbase/commit/4b3b627a Tree: http://git-wip-us.apache.org/repos/asf/hbase/tree/4b3b627a Diff: http://git-wip-us.apache.org/repos/asf/hbase/diff/4b3b627a Branch: refs/heads/HBASE-19064 Commit: 4b3b627abe34f8426fddd914e941d12f79da0752 Parents: 7c318ce Author: Rahul Gidwani <chu...@apache.org> Authored: Thu Jan 25 12:47:50 2018 -0800 Committer: Rahul Gidwani <chu...@apache.org> Committed: Wed Jan 31 10:18:03 2018 -0800 ---------------------------------------------------------------------- .../compaction/ClusterCompactionQueues.java | 137 +++++++ .../util/compaction/MajorCompactionRequest.java | 171 +++++++++ .../hbase/util/compaction/MajorCompactor.java | 379 +++++++++++++++++++ .../compaction/MajorCompactionRequestTest.java | 166 ++++++++ .../util/compaction/MajorCompactorTest.java | 81 ++++ 5 files changed, 934 insertions(+) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hbase/blob/4b3b627a/hbase-server/src/main/java/org/apache/hadoop/hbase/util/compaction/ClusterCompactionQueues.java ---------------------------------------------------------------------- diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/util/compaction/ClusterCompactionQueues.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/util/compaction/ClusterCompactionQueues.java new file mode 100644 index 0000000..c0d34d9 --- /dev/null +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/util/compaction/ClusterCompactionQueues.java @@ -0,0 +1,137 @@ +/** + * 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.util.compaction; + +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.locks.ReadWriteLock; +import java.util.concurrent.locks.ReentrantReadWriteLock; +import org.apache.hadoop.hbase.ServerName; +import org.apache.yetus.audience.InterfaceAudience; +import org.apache.hbase.thirdparty.com.google.common.annotations.VisibleForTesting; +import org.apache.hbase.thirdparty.com.google.common.collect.Lists; +import org.apache.hbase.thirdparty.com.google.common.collect.Maps; +import org.apache.hbase.thirdparty.com.google.common.collect.Sets; + +@InterfaceAudience.Private +class ClusterCompactionQueues { + + private final Map<ServerName, List<MajorCompactionRequest>> compactionQueues; + private final Set<ServerName> compactingServers; + private final ReadWriteLock lock; + private final int concurrentServers; + + ClusterCompactionQueues(int concurrentServers) { + this.concurrentServers = concurrentServers; + + this.compactionQueues = Maps.newHashMap(); + this.lock = new ReentrantReadWriteLock(); + this.compactingServers = Sets.newHashSet(); + } + + void addToCompactionQueue(ServerName serverName, MajorCompactionRequest info) { + this.lock.writeLock().lock(); + try { + List<MajorCompactionRequest> result = this.compactionQueues.get(serverName); + if (result == null) { + result = Lists.newArrayList(); + compactionQueues.put(serverName, result); + } + result.add(info); + } finally { + this.lock.writeLock().unlock(); + } + } + + boolean hasWorkItems() { + lock.readLock().lock(); + try { + return !this.compactionQueues.values().stream().allMatch(List::isEmpty); + } finally { + lock.readLock().unlock(); + } + } + + int getCompactionRequestsLeftToFinish() { + lock.readLock().lock(); + try { + int size = 0; + for (List<MajorCompactionRequest> queue : compactionQueues.values()) { + size += queue.size(); + } + return size; + } finally { + lock.readLock().unlock(); + } + } + + @VisibleForTesting List<MajorCompactionRequest> getQueue(ServerName serverName) { + lock.readLock().lock(); + try { + return compactionQueues.get(serverName); + } finally { + lock.readLock().unlock(); + } + } + + MajorCompactionRequest reserveForCompaction(ServerName serverName) { + lock.writeLock().lock(); + try { + if (!compactionQueues.get(serverName).isEmpty()) { + compactingServers.add(serverName); + return compactionQueues.get(serverName).remove(0); + } + return null; + } finally { + lock.writeLock().unlock(); + } + } + + void releaseCompaction(ServerName serverName) { + lock.writeLock().lock(); + try { + compactingServers.remove(serverName); + } finally { + lock.writeLock().unlock(); + } + } + + boolean atCapacity() { + lock.readLock().lock(); + try { + return compactingServers.size() >= concurrentServers; + } finally { + lock.readLock().unlock(); + } + } + + Optional<ServerName> getLargestQueueFromServersNotCompacting() { + lock.readLock().lock(); + try { + return compactionQueues.entrySet().stream() + .filter(entry -> !compactingServers.contains(entry.getKey())) + .max(Map.Entry.comparingByValue( + (o1, o2) -> Integer.compare(o1.size(), o2.size()))).map(Map.Entry::getKey); + } finally { + lock.readLock().unlock(); + } + } + +} http://git-wip-us.apache.org/repos/asf/hbase/blob/4b3b627a/hbase-server/src/main/java/org/apache/hadoop/hbase/util/compaction/MajorCompactionRequest.java ---------------------------------------------------------------------- diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/util/compaction/MajorCompactionRequest.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/util/compaction/MajorCompactionRequest.java new file mode 100644 index 0000000..51b2b9d --- /dev/null +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/util/compaction/MajorCompactionRequest.java @@ -0,0 +1,171 @@ +/** + * 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.util.compaction; + +import java.io.IOException; +import java.util.Collection; +import java.util.List; +import java.util.Optional; +import java.util.Set; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileStatus; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hbase.client.Admin; +import org.apache.hadoop.hbase.client.Connection; +import org.apache.hadoop.hbase.client.ConnectionFactory; +import org.apache.hadoop.hbase.client.RegionInfo; +import org.apache.hadoop.hbase.regionserver.HRegionFileSystem; +import org.apache.hadoop.hbase.regionserver.StoreFileInfo; +import org.apache.hadoop.hbase.util.FSUtils; +import org.apache.yetus.audience.InterfaceAudience; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.apache.hbase.thirdparty.com.google.common.annotations.VisibleForTesting; +import org.apache.hbase.thirdparty.com.google.common.collect.Sets; + +@InterfaceAudience.Private +class MajorCompactionRequest { + + private static final Logger LOG = LoggerFactory.getLogger(MajorCompactionRequest.class); + + private final Configuration configuration; + private final RegionInfo region; + private Set<String> stores; + private final long timestamp; + + @VisibleForTesting + MajorCompactionRequest(Configuration configuration, RegionInfo region, + Set<String> stores, long timestamp) { + this.configuration = configuration; + this.region = region; + this.stores = stores; + this.timestamp = timestamp; + } + + static Optional<MajorCompactionRequest> newRequest(Configuration configuration, RegionInfo info, + Set<String> stores, long timestamp) throws IOException { + MajorCompactionRequest request = + new MajorCompactionRequest(configuration, info, stores, timestamp); + return request.createRequest(configuration, stores); + } + + RegionInfo getRegion() { + return region; + } + + Set<String> getStores() { + return stores; + } + + void setStores(Set<String> stores) { + this.stores = stores; + } + + @VisibleForTesting + Optional<MajorCompactionRequest> createRequest(Configuration configuration, + Set<String> stores) throws IOException { + Set<String> familiesToCompact = getStoresRequiringCompaction(stores); + MajorCompactionRequest request = null; + if (!familiesToCompact.isEmpty()) { + request = new MajorCompactionRequest(configuration, region, familiesToCompact, timestamp); + } + return Optional.ofNullable(request); + } + + Set<String> getStoresRequiringCompaction(Set<String> requestedStores) throws IOException { + try(Connection connection = getConnection(configuration)) { + HRegionFileSystem fileSystem = getFileSystem(connection); + Set<String> familiesToCompact = Sets.newHashSet(); + for (String family : requestedStores) { + // do we have any store files? + Collection<StoreFileInfo> storeFiles = fileSystem.getStoreFiles(family); + if (storeFiles == null) { + LOG.info("Excluding store: " + family + " for compaction for region: " + fileSystem + .getRegionInfo().getEncodedName(), " has no store files"); + continue; + } + // check for reference files + if (fileSystem.hasReferences(family) && familyHasReferenceFile(fileSystem, family)) { + familiesToCompact.add(family); + LOG.info("Including store: " + family + " with: " + storeFiles.size() + + " files for compaction for region: " + fileSystem.getRegionInfo().getEncodedName()); + continue; + } + // check store file timestamps + boolean includeStore = false; + for (StoreFileInfo storeFile : storeFiles) { + if (storeFile.getModificationTime() < timestamp) { + LOG.info("Including store: " + family + " with: " + storeFiles.size() + + " files for compaction for region: " + + fileSystem.getRegionInfo().getEncodedName()); + familiesToCompact.add(family); + includeStore = true; + break; + } + } + if (!includeStore) { + LOG.info("Excluding store: " + family + " for compaction for region: " + fileSystem + .getRegionInfo().getEncodedName(), " already compacted"); + } + } + return familiesToCompact; + } + } + + @VisibleForTesting + Connection getConnection(Configuration configuration) throws IOException { + return ConnectionFactory.createConnection(configuration); + } + + private boolean familyHasReferenceFile(HRegionFileSystem fileSystem, String family) + throws IOException { + List<Path> referenceFiles = + getReferenceFilePaths(fileSystem.getFileSystem(), fileSystem.getStoreDir(family)); + for (Path referenceFile : referenceFiles) { + FileStatus status = fileSystem.getFileSystem().getFileLinkStatus(referenceFile); + if (status.getModificationTime() < timestamp) { + LOG.info("Including store: " + family + " for compaction for region: " + fileSystem + .getRegionInfo().getEncodedName() + " (reference store files)"); + return true; + } + } + return false; + + } + + @VisibleForTesting + List<Path> getReferenceFilePaths(FileSystem fileSystem, Path familyDir) + throws IOException { + return FSUtils.getReferenceFilePaths(fileSystem, familyDir); + } + + @VisibleForTesting + HRegionFileSystem getFileSystem(Connection connection) throws IOException { + Admin admin = connection.getAdmin(); + return HRegionFileSystem.openRegionFromFileSystem(admin.getConfiguration(), + FSUtils.getCurrentFileSystem(admin.getConfiguration()), + FSUtils.getTableDir(FSUtils.getRootDir(admin.getConfiguration()), region.getTable()), + region, true); + } + + @Override + public String toString() { + return "region: " + region.getEncodedName() + " store(s): " + stores; + } +} http://git-wip-us.apache.org/repos/asf/hbase/blob/4b3b627a/hbase-server/src/main/java/org/apache/hadoop/hbase/util/compaction/MajorCompactor.java ---------------------------------------------------------------------- diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/util/compaction/MajorCompactor.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/util/compaction/MajorCompactor.java new file mode 100644 index 0000000..c3372bb --- /dev/null +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/util/compaction/MajorCompactor.java @@ -0,0 +1,379 @@ +/** + * 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.util.compaction; + +import java.io.IOException; +import java.util.Arrays; +import java.util.List; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import org.apache.commons.cli.CommandLine; +import org.apache.commons.cli.CommandLineParser; +import org.apache.commons.cli.DefaultParser; +import org.apache.commons.cli.HelpFormatter; +import org.apache.commons.cli.Option; +import org.apache.commons.cli.Options; +import org.apache.commons.cli.ParseException; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hbase.HBaseConfiguration; +import org.apache.hadoop.hbase.HBaseInterfaceAudience; +import org.apache.hadoop.hbase.HConstants; +import org.apache.hadoop.hbase.HRegionLocation; +import org.apache.hadoop.hbase.NotServingRegionException; +import org.apache.hadoop.hbase.ServerName; +import org.apache.hadoop.hbase.TableName; +import org.apache.hadoop.hbase.client.Admin; +import org.apache.hadoop.hbase.client.CompactionState; +import org.apache.hadoop.hbase.client.Connection; +import org.apache.hadoop.hbase.client.ConnectionFactory; +import org.apache.hadoop.hbase.util.Bytes; +import org.apache.hbase.thirdparty.com.google.common.annotations.VisibleForTesting; +import org.apache.yetus.audience.InterfaceAudience; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.apache.hbase.thirdparty.com.google.common.base.Joiner; +import org.apache.hbase.thirdparty.com.google.common.base.Splitter; +import org.apache.hbase.thirdparty.com.google.common.collect.Iterables; +import org.apache.hbase.thirdparty.com.google.common.collect.Lists; +import org.apache.hbase.thirdparty.com.google.common.collect.Sets; + +@InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.TOOLS) +public class MajorCompactor { + + private static final Logger LOG = LoggerFactory.getLogger(MajorCompactor.class); + private static final Set<MajorCompactionRequest> ERRORS = ConcurrentHashMap.newKeySet(); + + private final ClusterCompactionQueues clusterCompactionQueues; + private final long timestamp; + private final Set<String> storesToCompact; + private final ExecutorService executor; + private final long sleepForMs; + private final Connection connection; + private final TableName tableName; + + public MajorCompactor(Configuration conf, TableName tableName, Set<String> storesToCompact, + int concurrency, long timestamp, long sleepForMs) throws IOException { + this.connection = ConnectionFactory.createConnection(conf); + this.tableName = tableName; + this.timestamp = timestamp; + this.storesToCompact = storesToCompact; + this.executor = Executors.newFixedThreadPool(concurrency); + this.clusterCompactionQueues = new ClusterCompactionQueues(concurrency); + this.sleepForMs = sleepForMs; + } + + public void compactAllRegions() throws Exception { + List<Future<?>> futures = Lists.newArrayList(); + while (clusterCompactionQueues.hasWorkItems() || !futuresComplete(futures)) { + while (clusterCompactionQueues.atCapacity()) { + LOG.debug("Waiting for servers to complete Compactions"); + Thread.sleep(sleepForMs); + } + Optional<ServerName> serverToProcess = + clusterCompactionQueues.getLargestQueueFromServersNotCompacting(); + if (serverToProcess.isPresent() && clusterCompactionQueues.hasWorkItems()) { + ServerName serverName = serverToProcess.get(); + // check to see if the region has moved... if so we have to enqueue it again with + // the proper serverName + MajorCompactionRequest request = clusterCompactionQueues.reserveForCompaction(serverName); + + ServerName currentServer = connection.getRegionLocator(tableName) + .getRegionLocation(request.getRegion().getStartKey()).getServerName(); + + if (!currentServer.equals(serverName)) { + // add it back to the queue with the correct server it should be picked up in the future. + LOG.info("Server changed for region: " + request.getRegion().getEncodedName() + " from: " + + serverName + " to: " + currentServer + " re-queuing request"); + clusterCompactionQueues.addToCompactionQueue(currentServer, request); + clusterCompactionQueues.releaseCompaction(serverName); + } else { + LOG.info("Firing off compaction request for server: " + serverName + ", " + request + + " total queue size left: " + clusterCompactionQueues + .getCompactionRequestsLeftToFinish()); + futures.add(executor.submit(new Compact(serverName, request))); + } + } else { + // haven't assigned anything so we sleep. + Thread.sleep(sleepForMs); + } + } + LOG.info("All compactions have completed"); + } + + private boolean futuresComplete(List<Future<?>> futures) { + futures.removeIf(Future::isDone); + return futures.isEmpty(); + } + + public void shutdown() throws Exception { + executor.shutdown(); + executor.awaitTermination(Long.MAX_VALUE, TimeUnit.MILLISECONDS); + if (!ERRORS.isEmpty()) { + StringBuilder builder = + new StringBuilder().append("Major compaction failed, there were: ").append(ERRORS.size()) + .append(" regions / stores that failed compacting\n") + .append("Failed compaction requests\n").append("--------------------------\n") + .append(Joiner.on("\n").join(ERRORS)); + LOG.error(builder.toString()); + } + if (connection != null) { + connection.close(); + } + LOG.info("All regions major compacted successfully"); + } + + @VisibleForTesting void initializeWorkQueues() throws IOException { + if (storesToCompact.isEmpty()) { + connection.getTable(tableName).getDescriptor().getColumnFamilyNames() + .forEach(a -> storesToCompact.add(Bytes.toString(a))); + LOG.info("No family specified, will execute for all families"); + } + LOG.info( + "Initializing compaction queues for table: " + tableName + " with cf: " + storesToCompact); + List<HRegionLocation> regionLocations = + connection.getRegionLocator(tableName).getAllRegionLocations(); + for (HRegionLocation location : regionLocations) { + Optional<MajorCompactionRequest> request = MajorCompactionRequest + .newRequest(connection.getConfiguration(), location.getRegion(), storesToCompact, + timestamp); + request.ifPresent(majorCompactionRequest -> clusterCompactionQueues + .addToCompactionQueue(location.getServerName(), majorCompactionRequest)); + } + } + + class Compact implements Runnable { + + private final ServerName serverName; + private final MajorCompactionRequest request; + + Compact(ServerName serverName, MajorCompactionRequest request) { + this.serverName = serverName; + this.request = request; + } + + @Override public void run() { + try { + compactAndWait(request); + } catch (NotServingRegionException e) { + // this region has split or merged + LOG.warn("Region is invalid, requesting updated regions", e); + // lets updated the cluster compaction queues with these newly created regions. + addNewRegions(); + } catch (Exception e) { + LOG.warn("Error compacting:", e); + } finally { + clusterCompactionQueues.releaseCompaction(serverName); + } + } + + void compactAndWait(MajorCompactionRequest request) throws Exception { + Admin admin = connection.getAdmin(); + try { + // only make the request if the region is not already major compacting + if (!isCompacting(request)) { + Set<String> stores = request.getStoresRequiringCompaction(storesToCompact); + if (!stores.isEmpty()) { + request.setStores(stores); + for (String store : request.getStores()) { + admin.majorCompactRegion(request.getRegion().getEncodedNameAsBytes(), + Bytes.toBytes(store)); + } + } + } + while (isCompacting(request)) { + Thread.sleep(sleepForMs); + LOG.debug("Waiting for compaction to complete for region: " + request.getRegion() + .getEncodedName()); + } + } finally { + // Make sure to wait for the CompactedFileDischarger chore to do its work + int waitForArchive = connection.getConfiguration() + .getInt("hbase.hfile.compaction.discharger.interval", 2 * 60 * 1000); + Thread.sleep(waitForArchive); + // check if compaction completed successfully, otherwise put that request back in the + // proper queue + Set<String> storesRequiringCompaction = + request.getStoresRequiringCompaction(storesToCompact); + if (!storesRequiringCompaction.isEmpty()) { + // this happens, when a region server is marked as dead, flushes a store file and + // the new regionserver doesn't pick it up because its accounted for in the WAL replay, + // thus you have more store files on the filesystem than the regionserver knows about. + boolean regionHasNotMoved = connection.getRegionLocator(tableName) + .getRegionLocation(request.getRegion().getStartKey()).getServerName() + .equals(serverName); + if (regionHasNotMoved) { + LOG.error("Not all store files were compacted, this may be due to the regionserver not " + + "being aware of all store files. Will not reattempt compacting, " + request); + ERRORS.add(request); + } else { + request.setStores(storesRequiringCompaction); + clusterCompactionQueues.addToCompactionQueue(serverName, request); + LOG.info("Compaction failed for the following stores: " + storesRequiringCompaction + + " region: " + request.getRegion().getEncodedName()); + } + } else { + LOG.info("Compaction complete for region: " + request.getRegion().getEncodedName() + + " -> cf(s): " + request.getStores()); + } + } + } + } + + private boolean isCompacting(MajorCompactionRequest request) throws Exception { + CompactionState compactionState = connection.getAdmin() + .getCompactionStateForRegion(request.getRegion().getEncodedNameAsBytes()); + return compactionState.equals(CompactionState.MAJOR) || compactionState + .equals(CompactionState.MAJOR_AND_MINOR); + } + + private void addNewRegions() { + try { + List<HRegionLocation> locations = + connection.getRegionLocator(tableName).getAllRegionLocations(); + for (HRegionLocation location : locations) { + if (location.getRegion().getRegionId() > timestamp) { + Optional<MajorCompactionRequest> compactionRequest = MajorCompactionRequest + .newRequest(connection.getConfiguration(), location.getRegion(), storesToCompact, + timestamp); + compactionRequest.ifPresent(request -> clusterCompactionQueues + .addToCompactionQueue(location.getServerName(), request)); + } + } + } catch (IOException e) { + throw new RuntimeException(e); + } + } + + public static void main(String[] args) throws Exception { + Options options = new Options(); + options.addOption( + Option.builder("table") + .required() + .desc("table name") + .hasArg() + .build() + ); + options.addOption( + Option.builder("cf") + .optionalArg(true) + .desc("column families: comma separated eg: a,b,c") + .hasArg() + .build() + ); + options.addOption( + Option.builder("servers") + .required() + .desc("Concurrent servers compacting") + .hasArg() + .build() + ); + options.addOption( + Option.builder("minModTime"). + desc("Compact if store files have modification time < minModTime") + .hasArg() + .build() + ); + options.addOption( + Option.builder("zk") + .optionalArg(true) + .desc("zk quorum") + .hasArg() + .build() + ); + options.addOption( + Option.builder("rootDir") + .optionalArg(true) + .desc("hbase.rootDir") + .hasArg() + .build() + ); + options.addOption( + Option.builder("sleep") + .desc("Time to sleepForMs (ms) for checking compaction status per region and available " + + "work queues: default 30s") + .hasArg() + .build() + ); + options.addOption( + Option.builder("retries") + .desc("Max # of retries for a compaction request," + " defaults to 3") + .hasArg() + .build() + ); + options.addOption( + Option.builder("dryRun") + .desc("Dry run, will just output a list of regions that require compaction based on " + + "parameters passed") + .hasArg(false) + .build() + ); + + final CommandLineParser cmdLineParser = new DefaultParser(); + CommandLine commandLine = null; + try { + commandLine = cmdLineParser.parse(options, args); + } catch (ParseException parseException) { + System.out.println( + "ERROR: Unable to parse command-line arguments " + Arrays.toString(args) + " due to: " + + parseException); + printUsage(options); + + } + String tableName = commandLine.getOptionValue("table"); + String cf = commandLine.getOptionValue("cf", null); + Set<String> families = Sets.newHashSet(); + if (cf != null) { + Iterables.addAll(families, Splitter.on(",").split(cf)); + } + + + Configuration configuration = HBaseConfiguration.create(); + int concurrency = Integer.parseInt(commandLine.getOptionValue("servers")); + long minModTime = Long.parseLong( + commandLine.getOptionValue("minModTime", String.valueOf(System.currentTimeMillis()))); + String quorum = + commandLine.getOptionValue("zk", configuration.get(HConstants.ZOOKEEPER_QUORUM)); + String rootDir = commandLine.getOptionValue("rootDir", configuration.get(HConstants.HBASE_DIR)); + long sleep = Long.valueOf(commandLine.getOptionValue("sleep", Long.toString(30000))); + + configuration.set(HConstants.HBASE_DIR, rootDir); + configuration.set(HConstants.ZOOKEEPER_QUORUM, quorum); + + MajorCompactor compactor = + new MajorCompactor(configuration, TableName.valueOf(tableName), families, concurrency, + minModTime, sleep); + + compactor.initializeWorkQueues(); + if (!commandLine.hasOption("dryRun")) { + compactor.compactAllRegions(); + } + compactor.shutdown(); + } + + private static void printUsage(final Options options) { + String header = "\nUsage instructions\n\n"; + String footer = "\n"; + HelpFormatter formatter = new HelpFormatter(); + formatter.printHelp(MajorCompactor.class.getSimpleName(), header, options, footer, true); + } + +} http://git-wip-us.apache.org/repos/asf/hbase/blob/4b3b627a/hbase-server/src/test/java/org/apache/hadoop/hbase/util/compaction/MajorCompactionRequestTest.java ---------------------------------------------------------------------- diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/util/compaction/MajorCompactionRequestTest.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/util/compaction/MajorCompactionRequestTest.java new file mode 100644 index 0000000..c5ce4e3 --- /dev/null +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/util/compaction/MajorCompactionRequestTest.java @@ -0,0 +1,166 @@ +/** + * 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.util.compaction; + +import java.io.IOException; +import java.util.List; +import java.util.Optional; +import java.util.Set; +import java.util.stream.Collectors; +import org.apache.commons.lang.RandomStringUtils; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileStatus; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hbase.HBaseTestingUtility; +import org.apache.hadoop.hbase.TableName; +import org.apache.hadoop.hbase.client.Connection; +import org.apache.hadoop.hbase.client.RegionInfo; +import org.apache.hadoop.hbase.client.RegionInfoBuilder; +import org.apache.hadoop.hbase.client.TableDescriptor; +import org.apache.hadoop.hbase.regionserver.HRegion; +import org.apache.hadoop.hbase.regionserver.HRegionFileSystem; +import org.apache.hadoop.hbase.regionserver.StoreFileInfo; +import org.apache.hadoop.hbase.testclassification.SmallTests; +import org.apache.hadoop.hbase.util.Bytes; +import org.apache.hbase.thirdparty.com.google.common.collect.Iterables; +import org.apache.hbase.thirdparty.com.google.common.collect.Lists; +import org.apache.hbase.thirdparty.com.google.common.collect.Sets; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +import org.junit.Before; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.ArgumentMatchers.isA; +import static org.mockito.Matchers.anyString; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.when; + +@Category({SmallTests.class}) +public class MajorCompactionRequestTest { + + private static final HBaseTestingUtility UTILITY = new HBaseTestingUtility(); + private static final String FAMILY = "a"; + private Path rootRegionDir; + private Path regionStoreDir; + + @Before public void setUp() throws Exception { + rootRegionDir = UTILITY.getDataTestDirOnTestFS("MajorCompactionRequestTest"); + regionStoreDir = new Path(rootRegionDir, FAMILY); + } + + @Test public void testStoresNeedingCompaction() throws Exception { + // store files older than timestamp + List<StoreFileInfo> storeFiles = mockStoreFiles(regionStoreDir, 5, 10); + MajorCompactionRequest request = makeMockRequest(100, storeFiles, false); + Optional<MajorCompactionRequest> result = + request.createRequest(mock(Configuration.class), Sets.newHashSet(FAMILY)); + assertTrue(result.isPresent()); + + // store files newer than timestamp + storeFiles = mockStoreFiles(regionStoreDir, 5, 101); + request = makeMockRequest(100, storeFiles, false); + result = request.createRequest(mock(Configuration.class), Sets.newHashSet(FAMILY)); + assertFalse(result.isPresent()); + } + + @Test public void testIfWeHaveNewReferenceFilesButOldStoreFiles() throws Exception { + // this tests that reference files that are new, but have older timestamps for the files + // they reference still will get compacted. + TableName table = TableName.valueOf("MajorCompactorTest"); + TableDescriptor htd = UTILITY.createTableDescriptor(table, Bytes.toBytes(FAMILY)); + RegionInfo hri = RegionInfoBuilder.newBuilder(htd.getTableName()).build(); + HRegion region = + HBaseTestingUtility.createRegionAndWAL(hri, rootRegionDir, UTILITY.getConfiguration(), htd); + + Configuration configuration = mock(Configuration.class); + // the reference file timestamp is newer + List<StoreFileInfo> storeFiles = mockStoreFiles(regionStoreDir, 4, 101); + List<Path> paths = storeFiles.stream().map(StoreFileInfo::getPath).collect(Collectors.toList()); + // the files that are referenced are older, thus we still compact. + HRegionFileSystem fileSystem = + mockFileSystem(region.getRegionInfo(), true, storeFiles, 50); + MajorCompactionRequest majorCompactionRequest = spy(new MajorCompactionRequest(configuration, + region.getRegionInfo(), Sets.newHashSet(FAMILY), 100)); + doReturn(mock(Connection.class)).when(majorCompactionRequest).getConnection(eq(configuration)); + doReturn(paths).when(majorCompactionRequest).getReferenceFilePaths(any(FileSystem.class), + any(Path.class)); + doReturn(fileSystem).when(majorCompactionRequest).getFileSystem(any(Connection.class)); + Set<String> result = majorCompactionRequest.getStoresRequiringCompaction(Sets.newHashSet("a")); + assertEquals(FAMILY, Iterables.getOnlyElement(result)); + } + + private HRegionFileSystem mockFileSystem(RegionInfo info, boolean hasReferenceFiles, + List<StoreFileInfo> storeFiles) throws IOException { + long timestamp = storeFiles.stream().findFirst().get().getModificationTime(); + return mockFileSystem(info, hasReferenceFiles, storeFiles, timestamp); + } + + private HRegionFileSystem mockFileSystem(RegionInfo info, boolean hasReferenceFiles, + List<StoreFileInfo> storeFiles, long referenceFileTimestamp) throws IOException { + FileSystem fileSystem = mock(FileSystem.class); + if (hasReferenceFiles) { + FileStatus fileStatus = mock(FileStatus.class); + doReturn(referenceFileTimestamp).when(fileStatus).getModificationTime(); + doReturn(fileStatus).when(fileSystem).getFileLinkStatus(isA(Path.class)); + } + HRegionFileSystem mockSystem = mock(HRegionFileSystem.class); + doReturn(info).when(mockSystem).getRegionInfo(); + doReturn(regionStoreDir).when(mockSystem).getStoreDir(FAMILY); + doReturn(hasReferenceFiles).when(mockSystem).hasReferences(anyString()); + doReturn(storeFiles).when(mockSystem).getStoreFiles(anyString()); + doReturn(fileSystem).when(mockSystem).getFileSystem(); + return mockSystem; + } + + private List<StoreFileInfo> mockStoreFiles(Path regionStoreDir, int howMany, long timestamp) + throws IOException { + List<StoreFileInfo> infos = Lists.newArrayList(); + int i = 0; + while (i < howMany) { + StoreFileInfo storeFileInfo = mock(StoreFileInfo.class); + doReturn(timestamp).doReturn(timestamp).when(storeFileInfo).getModificationTime(); + doReturn(new Path(regionStoreDir, RandomStringUtils.randomAlphabetic(10))).when(storeFileInfo) + .getPath(); + infos.add(storeFileInfo); + i++; + } + return infos; + } + + private MajorCompactionRequest makeMockRequest(long timestamp, List<StoreFileInfo> storeFiles, + boolean references) throws IOException { + Configuration configuration = mock(Configuration.class); + RegionInfo regionInfo = mock(RegionInfo.class); + when(regionInfo.getEncodedName()).thenReturn("HBase"); + when(regionInfo.getTable()).thenReturn(TableName.valueOf("foo")); + MajorCompactionRequest request = + new MajorCompactionRequest(configuration, regionInfo, Sets.newHashSet("a"), timestamp); + MajorCompactionRequest spy = spy(request); + HRegionFileSystem fileSystem = mockFileSystem(regionInfo, references, storeFiles); + doReturn(fileSystem).when(spy).getFileSystem(isA(Connection.class)); + doReturn(mock(Connection.class)).when(spy).getConnection(eq(configuration)); + return spy; + } +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/hbase/blob/4b3b627a/hbase-server/src/test/java/org/apache/hadoop/hbase/util/compaction/MajorCompactorTest.java ---------------------------------------------------------------------- diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/util/compaction/MajorCompactorTest.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/util/compaction/MajorCompactorTest.java new file mode 100644 index 0000000..3fb37ec --- /dev/null +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/util/compaction/MajorCompactorTest.java @@ -0,0 +1,81 @@ +/** + * 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.util.compaction; + +import org.apache.hadoop.hbase.HBaseTestingUtility; +import org.apache.hadoop.hbase.TableName; +import org.apache.hadoop.hbase.client.Connection; +import org.apache.hadoop.hbase.client.Table; +import org.apache.hadoop.hbase.testclassification.MediumTests; +import org.apache.hadoop.hbase.testclassification.MiscTests; +import org.apache.hadoop.hbase.util.Bytes; +import org.apache.hbase.thirdparty.com.google.common.collect.Sets; +import org.junit.After; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import org.junit.Before; +import org.junit.Test; +import org.junit.experimental.categories.Category; + + +@Category({ MiscTests.class, MediumTests.class }) +public class MajorCompactorTest { + + public static final byte[] FAMILY = Bytes.toBytes("a"); + private HBaseTestingUtility utility; + + @Before public void setUp() throws Exception { + utility = new HBaseTestingUtility(); + utility.getConfiguration().setInt("hbase.hfile.compaction.discharger.interval", 10); + utility.startMiniCluster(); + } + + @After public void tearDown() throws Exception { + utility.shutdownMiniCluster(); + } + + @Test public void testCompactingATable() throws Exception { + TableName tableName = TableName.valueOf("MajorCompactorTest"); + utility.createMultiRegionTable(tableName, FAMILY, 5); + utility.waitTableAvailable(tableName); + Connection connection = utility.getConnection(); + Table table = connection.getTable(tableName); + // write data and flush multiple store files: + for (int i = 0; i < 5; i++) { + utility.loadRandomRows(table, FAMILY, 50, 100); + utility.flush(tableName); + } + table.close(); + int numberOfRegions = utility.getAdmin().getRegions(tableName).size(); + int numHFiles = utility.getNumHFiles(tableName, FAMILY); + // we should have a table with more store files than we would before we major compacted. + assertTrue(numberOfRegions < numHFiles); + + MajorCompactor compactor = + new MajorCompactor(utility.getConfiguration(), tableName, + Sets.newHashSet(Bytes.toString(FAMILY)), 1, System.currentTimeMillis(), 200); + compactor.initializeWorkQueues(); + compactor.compactAllRegions(); + compactor.shutdown(); + + // verify that the store has been completely major compacted. + numberOfRegions = utility.getAdmin().getRegions(tableName).size(); + numHFiles = utility.getNumHFiles(tableName, FAMILY); + assertEquals(numHFiles, numberOfRegions); + } +} \ No newline at end of file
