You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by ps...@apache.org on 2022/08/19 10:52:18 UTC
[hbase] 02/02: HBASE-26969:Eliminate MOB renames when SFT is enabled
This is an automated email from the ASF dual-hosted git repository.
psomogyi pushed a commit to branch branch-2
in repository https://gitbox.apache.org/repos/asf/hbase.git
commit d073404706bf2c796d78ff6f7157172558118b7a
Author: BukrosSzabolcs <sz...@cloudera.com>
AuthorDate: Tue Jun 21 11:18:55 2022 +0200
HBASE-26969:Eliminate MOB renames when SFT is enabled
Signed-off-by: Wellington Chevreuil <wc...@apache.org>
Signed-off-by: Peter Somogyi <ps...@apache.org>
closes #4712
---
.../hadoop/hbase/mob/DefaultMobStoreCompactor.java | 29 ++-
.../hadoop/hbase/mob/DefaultMobStoreFlusher.java | 16 +-
.../hadoop/hbase/mob/MobFileCleanerChore.java | 44 ++--
.../java/org/apache/hadoop/hbase/mob/MobUtils.java | 31 ++-
.../hadoop/hbase/mob/RSMobFileCleanerChore.java | 272 +++++++++++++++++++++
.../hadoop/hbase/regionserver/HMobStore.java | 39 ++-
.../hadoop/hbase/regionserver/HRegionServer.java | 15 ++
.../apache/hadoop/hbase/regionserver/HStore.java | 2 +-
.../hbase/regionserver/compactions/Compactor.java | 23 +-
.../hadoop/hbase/mob/FaultyMobStoreCompactor.java | 6 +-
.../org/apache/hadoop/hbase/mob/MobTestUtil.java | 22 ++
.../hbase/mob/TestDefaultMobStoreFlusher.java | 68 ++++--
.../hadoop/hbase/mob/TestMobCompactionBase.java | 218 -----------------
.../hadoop/hbase/mob/TestMobCompactionOptMode.java | 11 +-
.../mob/TestMobCompactionOptRegionBatchMode.java | 14 +-
.../hbase/mob/TestMobCompactionRegularMode.java | 68 ------
.../TestMobCompactionRegularRegionBatchMode.java | 13 +-
.../hbase/mob/TestMobCompactionWithDefaults.java | 55 +++--
.../hadoop/hbase/mob/TestMobFileCleanerChore.java | 33 ++-
.../hadoop/hbase/mob/TestMobStoreCompaction.java | 28 ++-
.../org/apache/hadoop/hbase/mob/TestMobUtils.java | 5 +
...erChore.java => TestRSMobFileCleanerChore.java} | 113 ++++++---
22 files changed, 692 insertions(+), 433 deletions(-)
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/mob/DefaultMobStoreCompactor.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/mob/DefaultMobStoreCompactor.java
index e93d87237cc..e46eb0f6209 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/mob/DefaultMobStoreCompactor.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/mob/DefaultMobStoreCompactor.java
@@ -294,20 +294,20 @@ public class DefaultMobStoreCompactor extends DefaultCompactor {
* @param cleanSeqId When true, remove seqId(used to be mvcc) value which is <=
* smallestReadPoint
* @param throughputController The compaction throughput controller.
- * @param major Is a major compaction.
- * @param numofFilesToCompact the number of files to compact
+ * @param request compaction request.
* @param progress Progress reporter.
* @return Whether compaction ended; false if it was interrupted for any reason.
*/
@Override
protected boolean performCompaction(FileDetails fd, InternalScanner scanner, CellSink writer,
long smallestReadPoint, boolean cleanSeqId, ThroughputController throughputController,
- boolean major, int numofFilesToCompact, CompactionProgress progress) throws IOException {
+ CompactionRequestImpl request, CompactionProgress progress) throws IOException {
long bytesWrittenProgressForLog = 0;
long bytesWrittenProgressForShippedCall = 0;
// Clear old mob references
mobRefSet.get().clear();
boolean isUserRequest = userRequest.get();
+ boolean major = request.isAllFiles();
boolean compactMOBs = major && isUserRequest;
boolean discardMobMiss = conf.getBoolean(MobConstants.MOB_UNSAFE_DISCARD_MISS_KEY,
MobConstants.DEFAULT_MOB_DISCARD_MISS);
@@ -351,12 +351,12 @@ public class DefaultMobStoreCompactor extends DefaultCompactor {
throughputController.start(compactionName);
KeyValueScanner kvs = (scanner instanceof KeyValueScanner) ? (KeyValueScanner) scanner : null;
long shippedCallSizeLimit =
- (long) numofFilesToCompact * this.store.getColumnFamilyDescriptor().getBlocksize();
+ (long) request.getFiles().size() * this.store.getColumnFamilyDescriptor().getBlocksize();
Cell mobCell = null;
try {
- mobFileWriter = newMobWriter(fd, major);
+ mobFileWriter = newMobWriter(fd, major, request.getWriterCreationTracker());
fileName = Bytes.toBytes(mobFileWriter.getPath().getName());
do {
@@ -426,7 +426,7 @@ public class DefaultMobStoreCompactor extends DefaultCompactor {
LOG.debug("Closing output MOB File, length={} file={}, store={}", len,
mobFileWriter.getPath().getName(), getStoreInfo());
commitOrAbortMobWriter(mobFileWriter, fd.maxSeqId, mobCells, major);
- mobFileWriter = newMobWriter(fd, major);
+ mobFileWriter = newMobWriter(fd, major, request.getWriterCreationTracker());
fileName = Bytes.toBytes(mobFileWriter.getPath().getName());
mobCells = 0;
}
@@ -470,7 +470,7 @@ public class DefaultMobStoreCompactor extends DefaultCompactor {
long len = mobFileWriter.getPos();
if (len > maxMobFileSize) {
commitOrAbortMobWriter(mobFileWriter, fd.maxSeqId, mobCells, major);
- mobFileWriter = newMobWriter(fd, major);
+ mobFileWriter = newMobWriter(fd, major, request.getWriterCreationTracker());
fileName = Bytes.toBytes(mobFileWriter.getPath().getName());
mobCells = 0;
}
@@ -522,7 +522,7 @@ public class DefaultMobStoreCompactor extends DefaultCompactor {
long len = mobFileWriter.getPos();
if (len > maxMobFileSize) {
commitOrAbortMobWriter(mobFileWriter, fd.maxSeqId, mobCells, major);
- mobFileWriter = newMobWriter(fd, major);
+ mobFileWriter = newMobWriter(fd, major, request.getWriterCreationTracker());
fileName = Bytes.toBytes(mobFileWriter.getPath().getName());
mobCells = 0;
}
@@ -608,11 +608,16 @@ public class DefaultMobStoreCompactor extends DefaultCompactor {
}
}
- private StoreFileWriter newMobWriter(FileDetails fd, boolean major) throws IOException {
+ private StoreFileWriter newMobWriter(FileDetails fd, boolean major,
+ Consumer<Path> writerCreationTracker) throws IOException {
try {
- StoreFileWriter mobFileWriter = mobStore.createWriterInTmp(new Date(fd.latestPutTs),
- fd.maxKeyCount, major ? majorCompactionCompression : minorCompactionCompression,
- store.getRegionInfo().getStartKey(), true);
+ StoreFileWriter mobFileWriter = mobStore.getStoreEngine().requireWritingToTmpDirFirst()
+ ? mobStore.createWriterInTmp(new Date(fd.latestPutTs), fd.maxKeyCount,
+ major ? majorCompactionCompression : minorCompactionCompression,
+ store.getRegionInfo().getStartKey(), true)
+ : mobStore.createWriter(new Date(fd.latestPutTs), fd.maxKeyCount,
+ major ? majorCompactionCompression : minorCompactionCompression,
+ store.getRegionInfo().getStartKey(), true, writerCreationTracker);
LOG.debug("New MOB writer created={} store={}", mobFileWriter.getPath().getName(),
getStoreInfo());
// Add reference we get for compact MOB
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/mob/DefaultMobStoreFlusher.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/mob/DefaultMobStoreFlusher.java
index a7f2ecdf242..6df17e58e22 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/mob/DefaultMobStoreFlusher.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/mob/DefaultMobStoreFlusher.java
@@ -127,7 +127,8 @@ public class DefaultMobStoreFlusher extends DefaultStoreFlusher {
try {
// It's a mob store, flush the cells in a mob way. This is the difference of flushing
// between a normal and a mob store.
- performMobFlush(snapshot, cacheFlushId, scanner, writer, status, throughputController);
+ performMobFlush(snapshot, cacheFlushId, scanner, writer, status, throughputController,
+ writerCreationTracker);
} catch (IOException ioe) {
e = ioe;
// throw the exception out
@@ -171,16 +172,21 @@ public class DefaultMobStoreFlusher extends DefaultStoreFlusher {
*/
protected void performMobFlush(MemStoreSnapshot snapshot, long cacheFlushId,
InternalScanner scanner, StoreFileWriter writer, MonitoredTask status,
- ThroughputController throughputController) throws IOException {
+ ThroughputController throughputController, Consumer<Path> writerCreationTracker)
+ throws IOException {
StoreFileWriter mobFileWriter = null;
int compactionKVMax =
conf.getInt(HConstants.COMPACTION_KV_MAX, HConstants.COMPACTION_KV_MAX_DEFAULT);
long mobCount = 0;
long mobSize = 0;
long time = snapshot.getTimeRangeTracker().getMax();
- mobFileWriter = mobStore.createWriterInTmp(new Date(time), snapshot.getCellsCount(),
- store.getColumnFamilyDescriptor().getCompressionType(), store.getRegionInfo().getStartKey(),
- false);
+ mobFileWriter = mobStore.getStoreEngine().requireWritingToTmpDirFirst()
+ ? mobStore.createWriterInTmp(new Date(time), snapshot.getCellsCount(),
+ store.getColumnFamilyDescriptor().getCompressionType(), store.getRegionInfo().getStartKey(),
+ false)
+ : mobStore.createWriter(new Date(time), snapshot.getCellsCount(),
+ store.getColumnFamilyDescriptor().getCompressionType(), store.getRegionInfo().getStartKey(),
+ false, writerCreationTracker);
// the target path is {tableName}/.mob/{cfName}/mobFiles
// the relative path is mobFiles
byte[] fileName = Bytes.toBytes(mobFileWriter.getPath().getName());
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/mob/MobFileCleanerChore.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/mob/MobFileCleanerChore.java
index e2c1f8961de..eb2201417b0 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/mob/MobFileCleanerChore.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/mob/MobFileCleanerChore.java
@@ -168,14 +168,15 @@ public class MobFileCleanerChore extends ScheduledChore {
maxCreationTimeToArchive, table);
}
+ FileSystem fs = FileSystem.get(conf);
+ Set<String> regionNames = new HashSet<>();
Path rootDir = CommonFSUtils.getRootDir(conf);
Path tableDir = CommonFSUtils.getTableDir(rootDir, table);
- // How safe is this call?
- List<Path> regionDirs = FSUtils.getRegionDirs(FileSystem.get(conf), tableDir);
+ List<Path> regionDirs = FSUtils.getRegionDirs(fs, tableDir);
Set<String> allActiveMobFileName = new HashSet<String>();
- FileSystem fs = FileSystem.get(conf);
for (Path regionPath : regionDirs) {
+ regionNames.add(regionPath.getName());
for (ColumnFamilyDescriptor hcd : list) {
String family = hcd.getNameAsString();
Path storePath = new Path(regionPath, family);
@@ -203,13 +204,26 @@ public class MobFileCleanerChore extends ScheduledChore {
for (Path pp : storeFiles) {
currentPath = pp;
LOG.trace("Store file: {}", pp);
- HStoreFile sf =
- new HStoreFile(fs, pp, conf, CacheConfig.DISABLED, BloomType.NONE, true);
- sf.initReader();
- byte[] mobRefData = sf.getMetadataValue(HStoreFile.MOB_FILE_REFS);
- byte[] bulkloadMarkerData = sf.getMetadataValue(HStoreFile.BULKLOAD_TASK_KEY);
- // close store file to avoid memory leaks
- sf.closeStoreFile(true);
+ HStoreFile sf = null;
+ byte[] mobRefData = null;
+ byte[] bulkloadMarkerData = null;
+ try {
+ sf = new HStoreFile(fs, pp, conf, CacheConfig.DISABLED, BloomType.NONE, true);
+ sf.initReader();
+ mobRefData = sf.getMetadataValue(HStoreFile.MOB_FILE_REFS);
+ bulkloadMarkerData = sf.getMetadataValue(HStoreFile.BULKLOAD_TASK_KEY);
+ // close store file to avoid memory leaks
+ sf.closeStoreFile(true);
+ } catch (IOException ex) {
+ // When FileBased SFT is active the store dir can contain corrupted or incomplete
+ // files. So read errors are expected. We just skip these files.
+ if (ex instanceof FileNotFoundException) {
+ throw ex;
+ }
+ LOG.debug("Failed to get mob data from file: {} due to error.", pp.toString(),
+ ex);
+ continue;
+ }
if (mobRefData == null) {
if (bulkloadMarkerData == null) {
LOG.warn("Found old store file with no MOB_FILE_REFS: {} - "
@@ -264,9 +278,11 @@ public class MobFileCleanerChore extends ScheduledChore {
while (rit.hasNext()) {
LocatedFileStatus lfs = rit.next();
Path p = lfs.getPath();
- if (!allActiveMobFileName.contains(p.getName())) {
- // MOB is not in a list of active references, but it can be too
- // fresh, skip it in this case
+ String[] mobParts = p.getName().split("_");
+ String regionName = mobParts[mobParts.length - 1];
+
+ if (!regionNames.contains(regionName)) {
+ // MOB belonged to a region no longer hosted
long creationTime = fs.getFileStatus(p).getModificationTime();
if (creationTime < maxCreationTimeToArchive) {
LOG.trace("Archiving MOB file {} creation time={}", p,
@@ -277,7 +293,7 @@ public class MobFileCleanerChore extends ScheduledChore {
fs.getFileStatus(p).getModificationTime());
}
} else {
- LOG.trace("Keeping active MOB file: {}", p);
+ LOG.trace("Keeping MOB file with existing region: {}", p);
}
}
LOG.info(" MOB Cleaner found {} files to archive for table={} family={}", toArchive.size(),
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/mob/MobUtils.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/mob/MobUtils.java
index a409c9f42c4..aab589cc13f 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/mob/MobUtils.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/mob/MobUtils.java
@@ -33,6 +33,7 @@ import java.util.Date;
import java.util.List;
import java.util.Optional;
import java.util.UUID;
+import java.util.function.Consumer;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
@@ -584,6 +585,33 @@ public final class MobUtils {
CacheConfig cacheConfig, Encryption.Context cryptoContext, ChecksumType checksumType,
int bytesPerChecksum, int blocksize, BloomType bloomType, boolean isCompaction)
throws IOException {
+ return createWriter(conf, fs, family, path, maxKeyCount, compression, cacheConfig,
+ cryptoContext, checksumType, bytesPerChecksum, blocksize, bloomType, isCompaction, null);
+ }
+
+ /**
+ * Creates a writer for the mob file in temp directory.
+ * @param conf The current configuration.
+ * @param fs The current file system.
+ * @param family The descriptor of the current column family.
+ * @param path The path for a temp directory.
+ * @param maxKeyCount The key count.
+ * @param compression The compression algorithm.
+ * @param cacheConfig The current cache config.
+ * @param cryptoContext The encryption context.
+ * @param checksumType The checksum type.
+ * @param bytesPerChecksum The bytes per checksum.
+ * @param blocksize The HFile block size.
+ * @param bloomType The bloom filter type.
+ * @param isCompaction If the writer is used in compaction.
+ * @param writerCreationTracker to track the current writer in the store
+ * @return The writer for the mob file.
+ */
+ public static StoreFileWriter createWriter(Configuration conf, FileSystem fs,
+ ColumnFamilyDescriptor family, Path path, long maxKeyCount, Compression.Algorithm compression,
+ CacheConfig cacheConfig, Encryption.Context cryptoContext, ChecksumType checksumType,
+ int bytesPerChecksum, int blocksize, BloomType bloomType, boolean isCompaction,
+ Consumer<Path> writerCreationTracker) throws IOException {
if (compression == null) {
compression = HFile.DEFAULT_COMPRESSION_ALGORITHM;
}
@@ -602,7 +630,8 @@ public final class MobUtils {
.withCreateTime(EnvironmentEdgeManager.currentTime()).build();
StoreFileWriter w = new StoreFileWriter.Builder(conf, writerCacheConf, fs).withFilePath(path)
- .withBloomType(bloomType).withMaxKeyCount(maxKeyCount).withFileContext(hFileContext).build();
+ .withBloomType(bloomType).withMaxKeyCount(maxKeyCount).withFileContext(hFileContext)
+ .withWriterCreationTracker(writerCreationTracker).build();
return w;
}
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/mob/RSMobFileCleanerChore.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/mob/RSMobFileCleanerChore.java
new file mode 100644
index 00000000000..06e34988733
--- /dev/null
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/mob/RSMobFileCleanerChore.java
@@ -0,0 +1,272 @@
+/*
+ * 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.mob;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.LocatedFileStatus;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.RemoteIterator;
+import org.apache.hadoop.hbase.ScheduledChore;
+import org.apache.hadoop.hbase.TableDescriptors;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.backup.HFileArchiver;
+import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
+import org.apache.hadoop.hbase.client.TableDescriptor;
+import org.apache.hadoop.hbase.regionserver.HRegion;
+import org.apache.hadoop.hbase.regionserver.HRegionServer;
+import org.apache.hadoop.hbase.regionserver.HStore;
+import org.apache.hadoop.hbase.regionserver.HStoreFile;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.util.CommonFSUtils;
+import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
+import org.apache.yetus.audience.InterfaceAudience;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hbase.thirdparty.com.google.common.collect.SetMultimap;
+
+/**
+ * The class RSMobFileCleanerChore for running cleaner regularly to remove the obsolete (files which
+ * have no active references to) mob files that were referenced from the current RS.
+ */
+@InterfaceAudience.Private
+public class RSMobFileCleanerChore extends ScheduledChore {
+
+ private static final Logger LOG = LoggerFactory.getLogger(RSMobFileCleanerChore.class);
+ private final HRegionServer rs;
+
+ public RSMobFileCleanerChore(HRegionServer rs) {
+ super(rs.getServerName() + "-MobFileCleanerChore", rs,
+ rs.getConfiguration().getInt(MobConstants.MOB_CLEANER_PERIOD,
+ MobConstants.DEFAULT_MOB_CLEANER_PERIOD),
+ Math.round(rs.getConfiguration().getInt(MobConstants.MOB_CLEANER_PERIOD,
+ MobConstants.DEFAULT_MOB_CLEANER_PERIOD)
+ * ((ThreadLocalRandom.current().nextDouble() + 0.5D))),
+ TimeUnit.SECONDS);
+ // to prevent a load spike on the fs the initial delay is modified by +/- 50%
+ this.rs = rs;
+ }
+
+ public RSMobFileCleanerChore() {
+ this.rs = null;
+ }
+
+ @Override
+ protected void chore() {
+
+ long minAgeToArchive = rs.getConfiguration().getLong(MobConstants.MIN_AGE_TO_ARCHIVE_KEY,
+ MobConstants.DEFAULT_MIN_AGE_TO_ARCHIVE);
+ // We check only those MOB files, which creation time is less
+ // than maxCreationTimeToArchive. This is a current time - 1h. 1 hour gap
+ // gives us full confidence that all corresponding store files will
+ // exist at the time cleaning procedure begins and will be examined.
+ // So, if MOB file creation time is greater than this maxTimeToArchive,
+ // this will be skipped and won't be archived.
+ long maxCreationTimeToArchive = EnvironmentEdgeManager.currentTime() - minAgeToArchive;
+
+ TableDescriptors htds = rs.getTableDescriptors();
+ try {
+ FileSystem fs = FileSystem.get(rs.getConfiguration());
+
+ Map<String, TableDescriptor> map = null;
+ try {
+ map = htds.getAll();
+ } catch (IOException e) {
+ LOG.error("MobFileCleanerChore failed", e);
+ return;
+ }
+ Map<String, Map<String, List<String>>> referencedMOBs = new HashMap<>();
+ for (TableDescriptor htd : map.values()) {
+ // Now clean obsolete files for a table
+ LOG.info("Cleaning obsolete MOB files from table={}", htd.getTableName());
+ List<ColumnFamilyDescriptor> list = MobUtils.getMobColumnFamilies(htd);
+ List<HRegion> regions = rs.getRegions(htd.getTableName());
+ for (HRegion region : regions) {
+ for (ColumnFamilyDescriptor hcd : list) {
+ HStore store = region.getStore(hcd.getName());
+ Collection<HStoreFile> sfs = store.getStorefiles();
+ Set<String> regionMobs = new HashSet<String>();
+ Path currentPath = null;
+ try {
+ // collectinng referenced MOBs
+ for (HStoreFile sf : sfs) {
+ currentPath = sf.getPath();
+ sf.initReader();
+ byte[] mobRefData = sf.getMetadataValue(HStoreFile.MOB_FILE_REFS);
+ byte[] bulkloadMarkerData = sf.getMetadataValue(HStoreFile.BULKLOAD_TASK_KEY);
+ // close store file to avoid memory leaks
+ sf.closeStoreFile(true);
+ if (mobRefData == null) {
+ if (bulkloadMarkerData == null) {
+ LOG.warn(
+ "Found old store file with no MOB_FILE_REFS: {} - "
+ + "can not proceed until all old files will be MOB-compacted.",
+ currentPath);
+ return;
+ } else {
+ LOG.debug("Skipping file without MOB references (bulkloaded file):{}",
+ currentPath);
+ continue;
+ }
+ }
+ // file may or may not have MOB references, but was created by the distributed
+ // mob compaction code.
+ try {
+ SetMultimap<TableName, String> mobs =
+ MobUtils.deserializeMobFileRefs(mobRefData).build();
+ LOG.debug("Found {} mob references for store={}", mobs.size(), sf);
+ LOG.trace("Specific mob references found for store={} : {}", sf, mobs);
+ regionMobs.addAll(mobs.values());
+ } catch (RuntimeException exception) {
+ throw new IOException("failure getting mob references for hfile " + sf,
+ exception);
+ }
+ }
+ // collecting files, MOB included currently being written
+ regionMobs.addAll(store.getStoreFilesBeingWritten().stream()
+ .map(path -> path.getName()).collect(Collectors.toList()));
+
+ referencedMOBs
+ .computeIfAbsent(hcd.getNameAsString(), cf -> new HashMap<String, List<String>>())
+ .computeIfAbsent(region.getRegionInfo().getEncodedName(), name -> new ArrayList<>())
+ .addAll(regionMobs);
+
+ } catch (FileNotFoundException e) {
+ LOG.warn(
+ "Missing file:{} Starting MOB cleaning cycle from the beginning" + " due to error",
+ currentPath, e);
+ regionMobs.clear();
+ continue;
+ } catch (IOException e) {
+ LOG.error("Failed to clean the obsolete mob files for table={}",
+ htd.getTableName().getNameAsString(), e);
+ }
+ }
+ }
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Found: {} active mob refs for table={}",
+ referencedMOBs.values().stream().map(inner -> inner.values())
+ .flatMap(lists -> lists.stream()).mapToInt(lists -> lists.size()).sum(),
+ htd.getTableName().getNameAsString());
+ }
+ if (LOG.isTraceEnabled()) {
+ referencedMOBs.values().stream().forEach(innerMap -> innerMap.values().stream()
+ .forEach(mobFileList -> mobFileList.stream().forEach(LOG::trace)));
+ }
+
+ // collect regions referencing MOB files belonging to the current rs
+ Set<String> regionsCovered = new HashSet<>();
+ referencedMOBs.values().stream()
+ .forEach(regionMap -> regionsCovered.addAll(regionMap.keySet()));
+
+ for (ColumnFamilyDescriptor hcd : list) {
+ List<Path> toArchive = new ArrayList<Path>();
+ String family = hcd.getNameAsString();
+ Path dir = MobUtils.getMobFamilyPath(rs.getConfiguration(), htd.getTableName(), family);
+ RemoteIterator<LocatedFileStatus> rit = fs.listLocatedStatus(dir);
+ while (rit.hasNext()) {
+ LocatedFileStatus lfs = rit.next();
+ Path p = lfs.getPath();
+ String[] mobParts = p.getName().split("_");
+ String regionName = mobParts[mobParts.length - 1];
+
+ // skip MOB files not belonging to a region assigned to the current rs
+ if (!regionsCovered.contains(regionName)) {
+ LOG.trace("MOB file does not belong to current rs: {}", p);
+ continue;
+ }
+
+ // check active or actively written mob files
+ Map<String, List<String>> cfMobs = referencedMOBs.get(hcd.getNameAsString());
+ if (
+ cfMobs != null && cfMobs.get(regionName) != null
+ && cfMobs.get(regionName).contains(p.getName())
+ ) {
+ LOG.trace("Keeping active MOB file: {}", p);
+ continue;
+ }
+
+ // MOB is not in a list of active references, but it can be too
+ // fresh, skip it in this case
+ long creationTime = fs.getFileStatus(p).getModificationTime();
+ if (creationTime < maxCreationTimeToArchive) {
+ LOG.trace("Archiving MOB file {} creation time={}", p,
+ (fs.getFileStatus(p).getModificationTime()));
+ toArchive.add(p);
+ } else {
+ LOG.trace("Skipping fresh file: {}. Creation time={}", p,
+ fs.getFileStatus(p).getModificationTime());
+ }
+
+ }
+ LOG.info(" MOB Cleaner found {} files to archive for table={} family={}",
+ toArchive.size(), htd.getTableName().getNameAsString(), family);
+ archiveMobFiles(rs.getConfiguration(), htd.getTableName(), family.getBytes(), toArchive);
+ LOG.info(" MOB Cleaner archived {} files, table={} family={}", toArchive.size(),
+ htd.getTableName().getNameAsString(), family);
+ }
+
+ LOG.info("Cleaning obsolete MOB files finished for table={}", htd.getTableName());
+
+ }
+ } catch (IOException e) {
+ LOG.error("MOB Cleaner failed when trying to access the file system", e);
+ }
+ }
+
+ /**
+ * Archives the mob files.
+ * @param conf The current configuration.
+ * @param tableName The table name.
+ * @param family The name of the column family.
+ * @param storeFiles The files to be archived.
+ * @throws IOException exception
+ */
+ public void archiveMobFiles(Configuration conf, TableName tableName, byte[] family,
+ List<Path> storeFiles) throws IOException {
+
+ if (storeFiles.size() == 0) {
+ // nothing to remove
+ LOG.debug("Skipping archiving old MOB files - no files found for table={} cf={}", tableName,
+ Bytes.toString(family));
+ return;
+ }
+ Path mobTableDir = CommonFSUtils.getTableDir(MobUtils.getMobHome(conf), tableName);
+ FileSystem fs = storeFiles.get(0).getFileSystem(conf);
+
+ for (Path p : storeFiles) {
+ LOG.debug("MOB Cleaner is archiving: {}", p);
+ HFileArchiver.archiveStoreFile(conf, fs, MobUtils.getMobRegionInfo(tableName), mobTableDir,
+ family, p);
+ }
+ }
+}
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HMobStore.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HMobStore.java
index ac69eb8d324..5992c5941e9 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HMobStore.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HMobStore.java
@@ -28,6 +28,7 @@ import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.Consumer;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
@@ -184,7 +185,27 @@ public class HMobStore extends HStore {
}
Path path = getTempDir();
return createWriterInTmp(MobUtils.formatDate(date), path, maxKeyCount, compression, startKey,
- isCompaction);
+ isCompaction, null);
+ }
+
+ /**
+ * Creates the writer for the mob file in the mob family directory.
+ * @param date The latest date of written cells.
+ * @param maxKeyCount The key count.
+ * @param compression The compression algorithm.
+ * @param startKey The start key.
+ * @param isCompaction If the writer is used in compaction.
+ * @return The writer for the mob file. n
+ */
+ public StoreFileWriter createWriter(Date date, long maxKeyCount,
+ Compression.Algorithm compression, byte[] startKey, boolean isCompaction,
+ Consumer<Path> writerCreationTracker) throws IOException {
+ if (startKey == null) {
+ startKey = HConstants.EMPTY_START_ROW;
+ }
+ Path path = getPath();
+ return createWriterInTmp(MobUtils.formatDate(date), path, maxKeyCount, compression, startKey,
+ isCompaction, writerCreationTracker);
}
/**
@@ -198,11 +219,13 @@ public class HMobStore extends HStore {
* @return The writer for the mob file. n
*/
public StoreFileWriter createWriterInTmp(String date, Path basePath, long maxKeyCount,
- Compression.Algorithm compression, byte[] startKey, boolean isCompaction) throws IOException {
+ Compression.Algorithm compression, byte[] startKey, boolean isCompaction,
+ Consumer<Path> writerCreationTracker) throws IOException {
MobFileName mobFileName =
MobFileName.create(startKey, date, UUID.randomUUID().toString().replaceAll("-", ""),
getHRegion().getRegionInfo().getEncodedName());
- return createWriterInTmp(mobFileName, basePath, maxKeyCount, compression, isCompaction);
+ return createWriterInTmp(mobFileName, basePath, maxKeyCount, compression, isCompaction,
+ writerCreationTracker);
}
/**
@@ -214,13 +237,15 @@ public class HMobStore extends HStore {
* @param isCompaction If the writer is used in compaction.
* @return The writer for the mob file. n
*/
+
public StoreFileWriter createWriterInTmp(MobFileName mobFileName, Path basePath, long maxKeyCount,
- Compression.Algorithm compression, boolean isCompaction) throws IOException {
+ Compression.Algorithm compression, boolean isCompaction, Consumer<Path> writerCreationTracker)
+ throws IOException {
return MobUtils.createWriter(conf, getFileSystem(), getColumnFamilyDescriptor(),
new Path(basePath, mobFileName.getFileName()), maxKeyCount, compression, getCacheConfig(),
getStoreContext().getEncryptionContext(), StoreUtils.getChecksumType(conf),
StoreUtils.getBytesPerChecksum(conf), getStoreContext().getBlockSize(), BloomType.NONE,
- isCompaction);
+ isCompaction, writerCreationTracker);
}
/**
@@ -234,6 +259,10 @@ public class HMobStore extends HStore {
}
Path dstPath = new Path(targetPath, sourceFile.getName());
validateMobFile(sourceFile);
+ if (sourceFile.equals(targetPath)) {
+ LOG.info("File is already in the destination dir: {}", sourceFile);
+ return;
+ }
LOG.info(" FLUSH Renaming flushed file from {} to {}", sourceFile, dstPath);
Path parent = dstPath.getParent();
if (!getFileSystem().exists(parent)) {
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java
index 5408685311e..6980003b6c1 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java
@@ -131,6 +131,7 @@ import org.apache.hadoop.hbase.master.HMaster;
import org.apache.hadoop.hbase.master.LoadBalancer;
import org.apache.hadoop.hbase.master.balancer.BaseLoadBalancer;
import org.apache.hadoop.hbase.mob.MobFileCache;
+import org.apache.hadoop.hbase.mob.RSMobFileCleanerChore;
import org.apache.hadoop.hbase.monitoring.TaskMonitor;
import org.apache.hadoop.hbase.namequeues.NamedQueueRecorder;
import org.apache.hadoop.hbase.namequeues.SlowLogTableOpsChore;
@@ -553,6 +554,8 @@ public class HRegionServer extends Thread
private BrokenStoreFileCleaner brokenStoreFileCleaner;
+ private RSMobFileCleanerChore rsMobFileCleanerChore;
+
@InterfaceAudience.Private
CompactedHFilesDischarger compactedFileDischarger;
@@ -2203,6 +2206,10 @@ public class HRegionServer extends Thread
choreService.scheduleChore(brokenStoreFileCleaner);
}
+ if (this.rsMobFileCleanerChore != null) {
+ choreService.scheduleChore(rsMobFileCleanerChore);
+ }
+
// Leases is not a Thread. Internally it runs a daemon thread. If it gets
// an unhandled exception, it will just exit.
Threads.setDaemonThreadRunning(this.leaseManager, getName() + ".leaseChecker",
@@ -2299,6 +2306,8 @@ public class HRegionServer extends Thread
new BrokenStoreFileCleaner((int) (brokenStoreFileCleanerDelay + jitterValue),
brokenStoreFileCleanerPeriod, this, conf, this);
+ this.rsMobFileCleanerChore = new RSMobFileCleanerChore(this);
+
registerConfigurationObservers();
}
@@ -2781,6 +2790,7 @@ public class HRegionServer extends Thread
shutdownChore(storefileRefresher);
shutdownChore(fsUtilizationChore);
shutdownChore(slowLogTableOpsChore);
+ shutdownChore(rsMobFileCleanerChore);
// cancel the remaining scheduled chores (in case we missed out any)
// TODO: cancel will not cleanup the chores, so we need make sure we do not miss any
choreService.shutdown();
@@ -4056,6 +4066,11 @@ public class HRegionServer extends Thread
return brokenStoreFileCleaner;
}
+ @InterfaceAudience.Private
+ public RSMobFileCleanerChore getRSMobFileCleanerChore() {
+ return rsMobFileCleanerChore;
+ }
+
RSSnapshotVerifier getRsSnapshotVerifier() {
return rsSnapshotVerifier;
}
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 b0dfa92336e..c698cfd8aba 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
@@ -2421,7 +2421,7 @@ public class HStore
* {@link BrokenStoreFileCleaner} to prevent deleting the these files as they are not present in
* SFT yet.
*/
- Set<Path> getStoreFilesBeingWritten() {
+ public Set<Path> getStoreFilesBeingWritten() {
return storeFileWriterCreationTrackers.stream().flatMap(t -> t.get().stream())
.collect(Collectors.toSet());
}
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/compactions/Compactor.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/compactions/Compactor.java
index 0de3eeb024d..566e708b06d 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/compactions/Compactor.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/compactions/Compactor.java
@@ -361,7 +361,7 @@ public abstract class Compactor<T extends CellSink> {
writer = sinkFactory.createWriter(scanner, fd, dropCache, request.isMajor(),
request.getWriterCreationTracker());
finished = performCompaction(fd, scanner, writer, smallestReadPoint, cleanSeqId,
- throughputController, request.isAllFiles(), request.getFiles().size(), progress);
+ throughputController, request, progress);
if (!finished) {
throw new InterruptedIOException("Aborting compaction of store " + store + " in region "
+ store.getRegionInfo().getRegionNameAsString() + " because it was interrupted.");
@@ -400,20 +400,19 @@ public abstract class Compactor<T extends CellSink> {
/**
* Performs the compaction.
- * @param fd FileDetails of cell sink writer
- * @param scanner Where to read from.
- * @param writer Where to write to.
- * @param smallestReadPoint Smallest read point.
- * @param cleanSeqId When true, remove seqId(used to be mvcc) value which is <=
- * smallestReadPoint
- * @param major Is a major compaction.
- * @param numofFilesToCompact the number of files to compact
- * @param progress Progress reporter.
+ * @param fd FileDetails of cell sink writer
+ * @param scanner Where to read from.
+ * @param writer Where to write to.
+ * @param smallestReadPoint Smallest read point.
+ * @param cleanSeqId When true, remove seqId(used to be mvcc) value which is <=
+ * smallestReadPoint
+ * @param request compaction request.
+ * @param progress Progress reporter.
* @return Whether compaction ended; false if it was interrupted for some reason.
*/
protected boolean performCompaction(FileDetails fd, InternalScanner scanner, CellSink writer,
long smallestReadPoint, boolean cleanSeqId, ThroughputController throughputController,
- boolean major, int numofFilesToCompact, CompactionProgress progress) throws IOException {
+ CompactionRequestImpl request, CompactionProgress progress) throws IOException {
assert writer instanceof ShipperListener;
long bytesWrittenProgressForLog = 0;
long bytesWrittenProgressForShippedCall = 0;
@@ -435,7 +434,7 @@ public abstract class Compactor<T extends CellSink> {
throughputController.start(compactionName);
KeyValueScanner kvs = (scanner instanceof KeyValueScanner) ? (KeyValueScanner) scanner : null;
long shippedCallSizeLimit =
- (long) numofFilesToCompact * this.store.getColumnFamilyDescriptor().getBlocksize();
+ (long) request.getFiles().size() * this.store.getColumnFamilyDescriptor().getBlocksize();
try {
do {
hasMore = scanner.next(cells, scannerContext);
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/FaultyMobStoreCompactor.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/FaultyMobStoreCompactor.java
index 324b897d9fd..d37cc605f6b 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/FaultyMobStoreCompactor.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/FaultyMobStoreCompactor.java
@@ -42,6 +42,7 @@ import org.apache.hadoop.hbase.regionserver.ShipperListener;
import org.apache.hadoop.hbase.regionserver.StoreFileWriter;
import org.apache.hadoop.hbase.regionserver.compactions.CloseChecker;
import org.apache.hadoop.hbase.regionserver.compactions.CompactionProgress;
+import org.apache.hadoop.hbase.regionserver.compactions.CompactionRequestImpl;
import org.apache.hadoop.hbase.regionserver.throttle.ThroughputControlUtil;
import org.apache.hadoop.hbase.regionserver.throttle.ThroughputController;
import org.apache.hadoop.hbase.util.Bytes;
@@ -91,8 +92,9 @@ public class FaultyMobStoreCompactor extends DefaultMobStoreCompactor {
@Override
protected boolean performCompaction(FileDetails fd, InternalScanner scanner, CellSink writer,
long smallestReadPoint, boolean cleanSeqId, ThroughputController throughputController,
- boolean major, int numofFilesToCompact, CompactionProgress progress) throws IOException {
+ CompactionRequestImpl request, CompactionProgress progress) throws IOException {
+ boolean major = request.isAllFiles();
totalCompactions.incrementAndGet();
if (major) {
totalMajorCompactions.incrementAndGet();
@@ -143,7 +145,7 @@ public class FaultyMobStoreCompactor extends DefaultMobStoreCompactor {
throughputController.start(compactionName);
KeyValueScanner kvs = (scanner instanceof KeyValueScanner) ? (KeyValueScanner) scanner : null;
long shippedCallSizeLimit =
- (long) numofFilesToCompact * this.store.getColumnFamilyDescriptor().getBlocksize();
+ (long) request.getFiles().size() * this.store.getColumnFamilyDescriptor().getBlocksize();
Cell mobCell = null;
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/MobTestUtil.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/MobTestUtil.java
index 54563e2eb90..2641ed227ae 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/MobTestUtil.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/MobTestUtil.java
@@ -18,17 +18,26 @@
package org.apache.hadoop.hbase.mob;
import java.io.IOException;
+import java.util.Date;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ThreadLocalRandom;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseTestingUtility;
import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
+import org.apache.hadoop.hbase.io.compress.Compression;
+import org.apache.hadoop.hbase.io.crypto.Encryption;
+import org.apache.hadoop.hbase.io.hfile.CacheConfig;
import org.apache.hadoop.hbase.regionserver.StoreFileWriter;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
@@ -112,4 +121,17 @@ public class MobTestUtil {
scan.setAttribute(MobConstants.MOB_SCAN_RAW, Bytes.toBytes(Boolean.TRUE));
return util.countRows(table, scan);
}
+
+ public static Path generateMOBFileForRegion(Configuration conf, TableName tableName,
+ ColumnFamilyDescriptor familyDescriptor, String regionName) throws IOException {
+ Date date = new Date();
+ String dateStr = MobUtils.formatDate(date);
+ FileSystem fs = FileSystem.get(conf);
+ Path cfMOBDir = MobUtils.getMobFamilyPath(conf, tableName, familyDescriptor.getNameAsString());
+ StoreFileWriter writer = MobUtils.createWriter(conf, fs, familyDescriptor, dateStr, cfMOBDir,
+ 1000L, Compression.Algorithm.NONE, "startKey", CacheConfig.DISABLED, Encryption.Context.NONE,
+ false, "");
+ writer.close();
+ return writer.getPath();
+ }
}
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/TestDefaultMobStoreFlusher.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/TestDefaultMobStoreFlusher.java
index a07a3b0ab2d..056afaf6fc4 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/TestDefaultMobStoreFlusher.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/TestDefaultMobStoreFlusher.java
@@ -17,30 +17,37 @@
*/
package org.apache.hadoop.hbase.mob;
+import java.util.Arrays;
+import java.util.Collection;
import java.util.List;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseClassTestRule;
import org.apache.hadoop.hbase.HBaseTestingUtility;
-import org.apache.hadoop.hbase.HColumnDescriptor;
-import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
+import org.apache.hadoop.hbase.client.TableDescriptor;
+import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
+import org.apache.hadoop.hbase.regionserver.storefiletracker.StoreFileTrackerFactory;
import org.apache.hadoop.hbase.testclassification.LargeTests;
import org.apache.hadoop.hbase.util.Bytes;
-import org.junit.AfterClass;
+import org.junit.After;
import org.junit.Assert;
-import org.junit.BeforeClass;
+import org.junit.Before;
import org.junit.ClassRule;
import org.junit.Rule;
import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.junit.rules.TestName;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+@RunWith(Parameterized.class)
@Category(LargeTests.class)
public class TestDefaultMobStoreFlusher {
@@ -60,41 +67,52 @@ public class TestDefaultMobStoreFlusher {
@Rule
public TestName name = new TestName();
- @BeforeClass
- public static void setUpBeforeClass() throws Exception {
+ protected Boolean useFileBasedSFT;
+
+ public TestDefaultMobStoreFlusher(Boolean useFileBasedSFT) {
+ this.useFileBasedSFT = useFileBasedSFT;
+ }
+
+ @Parameterized.Parameters
+ public static Collection<Boolean> data() {
+ Boolean[] data = { false, true };
+ return Arrays.asList(data);
+ }
+
+ @Before
+ public void setUpBefore() throws Exception {
+ if (useFileBasedSFT) {
+ TEST_UTIL.getConfiguration().set(StoreFileTrackerFactory.TRACKER_IMPL,
+ "org.apache.hadoop.hbase.regionserver.storefiletracker.FileBasedStoreFileTracker");
+ }
TEST_UTIL.startMiniCluster(1);
}
- @AfterClass
- public static void tearDownAfterClass() throws Exception {
+ @After
+ public void tearDownAfter() throws Exception {
TEST_UTIL.shutdownMiniCluster();
}
@Test
public void testFlushNonMobFile() throws Exception {
- final TableName tableName = TableName.valueOf(name.getMethodName());
- HTableDescriptor desc = new HTableDescriptor(tableName);
- HColumnDescriptor hcd = new HColumnDescriptor(family);
- hcd.setMaxVersions(4);
- desc.addFamily(hcd);
-
- testFlushFile(desc);
+ final TableName tableName = TableName.valueOf(TestMobUtils.getTableName(name));
+ TableDescriptor tableDescriptor = TableDescriptorBuilder.newBuilder(tableName)
+ .setColumnFamily(ColumnFamilyDescriptorBuilder.newBuilder(family).setMaxVersions(4).build())
+ .build();
+ testFlushFile(tableDescriptor);
}
@Test
public void testFlushMobFile() throws Exception {
- final TableName tableName = TableName.valueOf(name.getMethodName());
- HTableDescriptor desc = new HTableDescriptor(tableName);
- HColumnDescriptor hcd = new HColumnDescriptor(family);
- hcd.setMobEnabled(true);
- hcd.setMobThreshold(3L);
- hcd.setMaxVersions(4);
- desc.addFamily(hcd);
-
- testFlushFile(desc);
+ final TableName tableName = TableName.valueOf(TestMobUtils.getTableName(name));
+ TableDescriptor tableDescriptor = TableDescriptorBuilder.newBuilder(tableName)
+ .setColumnFamily(ColumnFamilyDescriptorBuilder.newBuilder(family).setMobEnabled(true)
+ .setMobThreshold(3L).setMaxVersions(4).build())
+ .build();
+ testFlushFile(tableDescriptor);
}
- private void testFlushFile(HTableDescriptor htd) throws Exception {
+ private void testFlushFile(TableDescriptor htd) throws Exception {
Table table = null;
try {
table = TEST_UTIL.createTable(htd, null);
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/TestMobCompactionBase.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/TestMobCompactionBase.java
deleted file mode 100644
index 62ca370197b..00000000000
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/TestMobCompactionBase.java
+++ /dev/null
@@ -1,218 +0,0 @@
-/*
- * 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.mob;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
-
-import java.io.IOException;
-import java.util.Arrays;
-import java.util.Random;
-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.HColumnDescriptor;
-import org.apache.hadoop.hbase.HTableDescriptor;
-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.Put;
-import org.apache.hadoop.hbase.client.Result;
-import org.apache.hadoop.hbase.client.ResultScanner;
-import org.apache.hadoop.hbase.client.Table;
-import org.apache.hadoop.hbase.util.Bytes;
-import org.apache.hadoop.hbase.util.RegionSplitter;
-import org.junit.After;
-import org.junit.Before;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Mob file compaction base test. 1. Enables batch mode for regular MOB compaction, Sets batch size
- * to 7 regions. (Optional) 2. Disables periodic MOB compactions, sets minimum age to archive to 10
- * sec 3. Creates MOB table with 20 regions 4. Loads MOB data (randomized keys, 1000 rows), flushes
- * data. 5. Repeats 4. two more times 6. Verifies that we have 20 *3 = 60 mob files (equals to
- * number of regions x 3) 7. Runs major MOB compaction. 8. Verifies that number of MOB files in a
- * mob directory is 20 x4 = 80 9. Waits for a period of time larger than minimum age to archive 10.
- * Runs Mob cleaner chore 11 Verifies that number of MOB files in a mob directory is 20. 12 Runs
- * scanner and checks all 3 * 1000 rows.
- */
-@SuppressWarnings("deprecation")
-public abstract class TestMobCompactionBase {
- private static final Logger LOG = LoggerFactory.getLogger(TestMobCompactionBase.class);
-
- protected HBaseTestingUtility HTU;
-
- protected final static String famStr = "f1";
- protected final static byte[] fam = Bytes.toBytes(famStr);
- protected final static byte[] qualifier = Bytes.toBytes("q1");
- protected final static long mobLen = 10;
- protected final static byte[] mobVal = Bytes
- .toBytes("01234567890123456789012345678901234567890123456789012345678901234567890123456789");
-
- protected Configuration conf;
- protected HTableDescriptor hdt;
- private HColumnDescriptor hcd;
- protected Admin admin;
- protected Table table = null;
- protected long minAgeToArchive = 10000;
- protected int numRegions = 20;
- protected int rows = 1000;
-
- protected MobFileCleanerChore cleanerChore;
-
- public TestMobCompactionBase() {
- }
-
- @Before
- public void setUp() throws Exception {
- HTU = new HBaseTestingUtility();
- hdt = HTU.createTableDescriptor(getClass().getName());
- conf = HTU.getConfiguration();
-
- initConf();
-
- HTU.startMiniCluster();
- admin = HTU.getAdmin();
- cleanerChore = new MobFileCleanerChore();
- hcd = new HColumnDescriptor(fam);
- hcd.setMobEnabled(true);
- hcd.setMobThreshold(mobLen);
- hcd.setMaxVersions(1);
- hdt.addFamily(hcd);
- RegionSplitter.UniformSplit splitAlgo = new RegionSplitter.UniformSplit();
- byte[][] splitKeys = splitAlgo.split(numRegions);
- table = HTU.createTable(hdt, splitKeys);
-
- }
-
- protected void initConf() {
-
- conf.setInt("hfile.format.version", 3);
- // Disable automatic MOB compaction
- conf.setLong(MobConstants.MOB_COMPACTION_CHORE_PERIOD, 0);
- // Disable automatic MOB file cleaner chore
- conf.setLong(MobConstants.MOB_CLEANER_PERIOD, 0);
- // Set minimum age to archive to 10 sec
- conf.setLong(MobConstants.MIN_AGE_TO_ARCHIVE_KEY, minAgeToArchive);
- // Set compacted file discharger interval to a half minAgeToArchive
- conf.setLong("hbase.hfile.compaction.discharger.interval", minAgeToArchive / 2);
- }
-
- private void loadData(int num) {
-
- Random r = new Random();
- try {
- LOG.info("Started loading {} rows", num);
- for (int i = 0; i < num; i++) {
- byte[] key = new byte[32];
- r.nextBytes(key);
- Put p = new Put(key);
- p.addColumn(fam, qualifier, mobVal);
- table.put(p);
- }
- admin.flush(table.getName());
- LOG.info("Finished loading {} rows", num);
- } catch (Exception e) {
- LOG.error("MOB file compaction chore test FAILED", e);
- fail("MOB file compaction chore test FAILED");
- }
- }
-
- @After
- public void tearDown() throws Exception {
- admin.disableTable(hdt.getTableName());
- admin.deleteTable(hdt.getTableName());
- HTU.shutdownMiniCluster();
- }
-
- public void baseTestMobFileCompaction() throws InterruptedException, IOException {
-
- // Load and flush data 3 times
- loadData(rows);
- loadData(rows);
- loadData(rows);
- long num = getNumberOfMobFiles(conf, table.getName(), new String(fam));
- assertEquals(numRegions * 3, num);
- // Major MOB compact
- mobCompact(admin, hdt, hcd);
- // wait until compaction is complete
- while (admin.getCompactionState(hdt.getTableName()) != CompactionState.NONE) {
- Thread.sleep(100);
- }
-
- num = getNumberOfMobFiles(conf, table.getName(), new String(fam));
- assertEquals(numRegions * 4, num);
- // We have guarantee, that compacted file discharger will run during this pause
- // because it has interval less than this wait time
- LOG.info("Waiting for {}ms", minAgeToArchive + 1000);
-
- Thread.sleep(minAgeToArchive + 1000);
- LOG.info("Cleaning up MOB files");
- // Cleanup again
- cleanerChore.cleanupObsoleteMobFiles(conf, table.getName());
-
- num = getNumberOfMobFiles(conf, table.getName(), new String(fam));
- assertEquals(numRegions, num);
-
- long scanned = scanTable();
- assertEquals(3 * rows, scanned);
-
- }
-
- protected abstract void mobCompact(Admin admin2, HTableDescriptor hdt2, HColumnDescriptor hcd2)
- throws IOException, InterruptedException;
-
- protected long getNumberOfMobFiles(Configuration conf, TableName tableName, String family)
- throws IOException {
- FileSystem fs = FileSystem.get(conf);
- Path dir = MobUtils.getMobFamilyPath(conf, tableName, family);
- FileStatus[] stat = fs.listStatus(dir);
- for (FileStatus st : stat) {
- LOG.debug("MOB Directory content: {}", st.getPath());
- }
- LOG.debug("MOB Directory content total files: {}", stat.length);
-
- return stat.length;
- }
-
- protected long scanTable() {
- try {
-
- Result result;
- ResultScanner scanner = table.getScanner(fam);
- long counter = 0;
- while ((result = scanner.next()) != null) {
- assertTrue(Arrays.equals(result.getValue(fam, qualifier), mobVal));
- counter++;
- }
- return counter;
- } catch (Exception e) {
- LOG.error("MOB file compaction test FAILED", e);
- if (HTU != null) {
- fail(e.getMessage());
- } else {
- System.exit(-1);
- }
- }
- return 0;
- }
-}
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/TestMobCompactionOptMode.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/TestMobCompactionOptMode.java
index 47fcde9a233..816ea29b911 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/TestMobCompactionOptMode.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/TestMobCompactionOptMode.java
@@ -17,10 +17,8 @@
*/
package org.apache.hadoop.hbase.mob;
-import java.io.IOException;
import org.apache.hadoop.hbase.HBaseClassTestRule;
import org.apache.hadoop.hbase.testclassification.LargeTests;
-import org.junit.BeforeClass;
import org.junit.ClassRule;
import org.junit.experimental.categories.Category;
@@ -40,12 +38,13 @@ public class TestMobCompactionOptMode extends TestMobCompactionWithDefaults {
public static final HBaseClassTestRule CLASS_RULE =
HBaseClassTestRule.forClass(TestMobCompactionOptMode.class);
- @BeforeClass
- public static void configureOptimizedCompaction() throws InterruptedException, IOException {
- HTU.shutdownMiniHBaseCluster();
+ public TestMobCompactionOptMode(Boolean useFileBasedSFT) {
+ super(useFileBasedSFT);
+ }
+
+ protected void additonalConfigSetup() {
conf.set(MobConstants.MOB_COMPACTION_TYPE_KEY, MobConstants.OPTIMIZED_MOB_COMPACTION_TYPE);
conf.setLong(MobConstants.MOB_COMPACTION_MAX_FILE_SIZE_KEY, 1000000);
- HTU.startMiniHBaseCluster();
}
@Override
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/TestMobCompactionOptRegionBatchMode.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/TestMobCompactionOptRegionBatchMode.java
index 7b6b44d0e31..dfd6435d364 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/TestMobCompactionOptRegionBatchMode.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/TestMobCompactionOptRegionBatchMode.java
@@ -23,9 +23,10 @@ import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
import org.apache.hadoop.hbase.client.TableDescriptor;
import org.apache.hadoop.hbase.testclassification.LargeTests;
import org.junit.Before;
-import org.junit.BeforeClass;
import org.junit.ClassRule;
import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -39,6 +40,7 @@ import org.slf4j.LoggerFactory;
* time larger than minimum age to archive 10. Runs Mob cleaner chore 11 Verifies that number of MOB
* files in a mob directory is 20. 12 Runs scanner and checks all 3 * 1000 rows.
*/
+@RunWith(Parameterized.class)
@Category(LargeTests.class)
public class TestMobCompactionOptRegionBatchMode extends TestMobCompactionWithDefaults {
private static final Logger LOG =
@@ -50,20 +52,20 @@ public class TestMobCompactionOptRegionBatchMode extends TestMobCompactionWithDe
private static final int batchSize = 7;
private MobFileCompactionChore compactionChore;
+ public TestMobCompactionOptRegionBatchMode(Boolean useFileBasedSFT) {
+ super(useFileBasedSFT);
+ }
+
@Before
public void setUp() throws Exception {
super.setUp();
compactionChore = new MobFileCompactionChore(conf, batchSize);
}
- @BeforeClass
- public static void configureOptimizedCompactionAndBatches()
- throws InterruptedException, IOException {
- HTU.shutdownMiniHBaseCluster();
+ protected void additonalConfigSetup() {
conf.setInt(MobConstants.MOB_MAJOR_COMPACTION_REGION_BATCH_SIZE, batchSize);
conf.set(MobConstants.MOB_COMPACTION_TYPE_KEY, MobConstants.OPTIMIZED_MOB_COMPACTION_TYPE);
conf.setLong(MobConstants.MOB_COMPACTION_MAX_FILE_SIZE_KEY, 1000000);
- HTU.startMiniHBaseCluster();
}
@Override
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/TestMobCompactionRegularMode.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/TestMobCompactionRegularMode.java
deleted file mode 100644
index 92809ea5511..00000000000
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/TestMobCompactionRegularMode.java
+++ /dev/null
@@ -1,68 +0,0 @@
-/*
- * 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.mob;
-
-import java.io.IOException;
-import org.apache.hadoop.hbase.HBaseClassTestRule;
-import org.apache.hadoop.hbase.HColumnDescriptor;
-import org.apache.hadoop.hbase.HTableDescriptor;
-import org.apache.hadoop.hbase.client.Admin;
-import org.apache.hadoop.hbase.testclassification.LargeTests;
-import org.junit.ClassRule;
-import org.junit.Test;
-import org.junit.experimental.categories.Category;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Mob file compaction chore in a regular non-batch mode test. 1. Uses default (non-batch) mode for
- * regular MOB compaction, 2. Disables periodic MOB compactions, sets minimum age to archive to 10
- * sec 3. Creates MOB table with 20 regions 4. Loads MOB data (randomized keys, 1000 rows), flushes
- * data. 5. Repeats 4. two more times 6. Verifies that we have 20 *3 = 60 mob files (equals to
- * number of regions x 3) 7. Runs major MOB compaction. 8. Verifies that number of MOB files in a
- * mob directory is 20 x4 = 80 9. Waits for a period of time larger than minimum age to archive 10.
- * Runs Mob cleaner chore 11 Verifies that number of MOB files in a mob directory is 20. 12 Runs
- * scanner and checks all 3 * 1000 rows.
- */
-@SuppressWarnings("deprecation")
-@Category(LargeTests.class)
-public class TestMobCompactionRegularMode extends TestMobCompactionBase {
- private static final Logger LOG = LoggerFactory.getLogger(TestMobCompactionRegularMode.class);
- @ClassRule
- public static final HBaseClassTestRule CLASS_RULE =
- HBaseClassTestRule.forClass(TestMobCompactionRegularMode.class);
-
- public TestMobCompactionRegularMode() {
- }
-
- @Test
- public void testMobFileCompactionBatchMode() throws InterruptedException, IOException {
- LOG.info("MOB compaction regular mode started");
- baseTestMobFileCompaction();
- LOG.info("MOB compaction regular mode finished OK");
-
- }
-
- @Override
- protected void mobCompact(Admin admin, HTableDescriptor hdt, HColumnDescriptor hcd)
- throws IOException, InterruptedException {
- // Major compact MOB table
- admin.majorCompact(hdt.getTableName(), hcd.getName());
- }
-
-}
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/TestMobCompactionRegularRegionBatchMode.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/TestMobCompactionRegularRegionBatchMode.java
index 3d6eaa0a25a..5e2806bdee2 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/TestMobCompactionRegularRegionBatchMode.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/TestMobCompactionRegularRegionBatchMode.java
@@ -23,9 +23,10 @@ import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
import org.apache.hadoop.hbase.client.TableDescriptor;
import org.apache.hadoop.hbase.testclassification.LargeTests;
import org.junit.Before;
-import org.junit.BeforeClass;
import org.junit.ClassRule;
import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -39,6 +40,7 @@ import org.slf4j.LoggerFactory;
* to archive 10. Runs Mob cleaner chore 11 Verifies that number of MOB files in a mob directory is
* 20. 12 Runs scanner and checks all 3 * 1000 rows.
*/
+@RunWith(Parameterized.class)
@Category(LargeTests.class)
public class TestMobCompactionRegularRegionBatchMode extends TestMobCompactionWithDefaults {
private static final Logger LOG =
@@ -50,17 +52,18 @@ public class TestMobCompactionRegularRegionBatchMode extends TestMobCompactionWi
private static final int batchSize = 7;
private MobFileCompactionChore compactionChore;
+ public TestMobCompactionRegularRegionBatchMode(Boolean useFileBasedSFT) {
+ super(useFileBasedSFT);
+ }
+
@Before
public void setUp() throws Exception {
super.setUp();
compactionChore = new MobFileCompactionChore(conf, batchSize);
}
- @BeforeClass
- public static void configureCompactionBatches() throws InterruptedException, IOException {
- HTU.shutdownMiniHBaseCluster();
+ protected void additonalConfigSetup() {
conf.setInt(MobConstants.MOB_MAJOR_COMPACTION_REGION_BATCH_SIZE, batchSize);
- HTU.startMiniHBaseCluster();
}
@Override
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/TestMobCompactionWithDefaults.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/TestMobCompactionWithDefaults.java
index 23e7c233496..df3eb29525b 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/TestMobCompactionWithDefaults.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/TestMobCompactionWithDefaults.java
@@ -23,6 +23,7 @@ import static org.junit.Assert.fail;
import java.io.IOException;
import java.util.Arrays;
+import java.util.Collection;
import java.util.List;
import java.util.Random;
import java.util.stream.Collectors;
@@ -32,6 +33,7 @@ import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseClassTestRule;
import org.apache.hadoop.hbase.HBaseTestingUtility;
+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.ColumnFamilyDescriptor;
@@ -43,18 +45,19 @@ import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.client.TableDescriptor;
import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
+import org.apache.hadoop.hbase.regionserver.storefiletracker.StoreFileTrackerFactory;
import org.apache.hadoop.hbase.testclassification.LargeTests;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.RegionSplitter;
import org.junit.After;
-import org.junit.AfterClass;
import org.junit.Before;
-import org.junit.BeforeClass;
import org.junit.ClassRule;
import org.junit.Rule;
import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.junit.rules.TestName;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -68,6 +71,7 @@ import org.slf4j.LoggerFactory;
* Runs Mob cleaner chore 11 Verifies that number of MOB files in a mob directory is 20. 12 Runs
* scanner and checks all 3 * 1000 rows.
*/
+@RunWith(Parameterized.class)
@Category(LargeTests.class)
public class TestMobCompactionWithDefaults {
private static final Logger LOG = LoggerFactory.getLogger(TestMobCompactionWithDefaults.class);
@@ -97,8 +101,19 @@ public class TestMobCompactionWithDefaults {
protected MobFileCleanerChore cleanerChore;
- @BeforeClass
- public static void htuStart() throws Exception {
+ protected Boolean useFileBasedSFT;
+
+ public TestMobCompactionWithDefaults(Boolean useFileBasedSFT) {
+ this.useFileBasedSFT = useFileBasedSFT;
+ }
+
+ @Parameterized.Parameters
+ public static Collection<Boolean> data() {
+ Boolean[] data = { false, true };
+ return Arrays.asList(data);
+ }
+
+ protected void htuStart() throws Exception {
HTU = new HBaseTestingUtility();
conf = HTU.getConfiguration();
conf.setInt("hfile.format.version", 3);
@@ -111,17 +126,21 @@ public class TestMobCompactionWithDefaults {
// Set compacted file discharger interval to a half minAgeToArchive
conf.setLong("hbase.hfile.compaction.discharger.interval", minAgeToArchive / 2);
conf.setBoolean("hbase.regionserver.compaction.enabled", false);
+ if (useFileBasedSFT) {
+ conf.set(StoreFileTrackerFactory.TRACKER_IMPL,
+ "org.apache.hadoop.hbase.regionserver.storefiletracker.FileBasedStoreFileTracker");
+ }
+ additonalConfigSetup();
HTU.startMiniCluster();
}
- @AfterClass
- public static void htuStop() throws Exception {
- HTU.shutdownMiniCluster();
+ protected void additonalConfigSetup() {
}
@Before
public void setUp() throws Exception {
- tableDescriptor = HTU.createModifyableTableDescriptor(test.getMethodName());
+ htuStart();
+ tableDescriptor = HTU.createModifyableTableDescriptor(TestMobUtils.getTableName(test));
admin = HTU.getAdmin();
cleanerChore = new MobFileCleanerChore();
familyDescriptor = new ColumnFamilyDescriptorBuilder.ModifyableColumnFamilyDescriptor(fam);
@@ -158,6 +177,7 @@ public class TestMobCompactionWithDefaults {
public void tearDown() throws Exception {
admin.disableTable(tableDescriptor.getTableName());
admin.deleteTable(tableDescriptor.getTableName());
+ HTU.shutdownMiniCluster();
}
@Test
@@ -173,12 +193,12 @@ public class TestMobCompactionWithDefaults {
@Test
public void testMobFileCompactionAfterSnapshotClone() throws InterruptedException, IOException {
- final TableName clone = TableName.valueOf(test.getMethodName() + "-clone");
+ final TableName clone = TableName.valueOf(TestMobUtils.getTableName(test) + "-clone");
LOG.info("MOB compaction of cloned snapshot, " + description() + " started");
loadAndFlushThreeTimes(rows, table, famStr);
LOG.debug("Taking snapshot and cloning table {}", table);
- admin.snapshot(test.getMethodName(), table);
- admin.cloneSnapshot(test.getMethodName(), clone);
+ admin.snapshot(TestMobUtils.getTableName(test), table);
+ admin.cloneSnapshot(TestMobUtils.getTableName(test), clone);
assertEquals("Should have 3 hlinks per region in MOB area from snapshot clone", 3 * numRegions,
getNumberOfMobFiles(clone, famStr));
mobCompact(admin.getDescriptor(clone), familyDescriptor);
@@ -191,12 +211,12 @@ public class TestMobCompactionWithDefaults {
@Test
public void testMobFileCompactionAfterSnapshotCloneAndFlush()
throws InterruptedException, IOException {
- final TableName clone = TableName.valueOf(test.getMethodName() + "-clone");
+ final TableName clone = TableName.valueOf(TestMobUtils.getTableName(test) + "-clone");
LOG.info("MOB compaction of cloned snapshot after flush, " + description() + " started");
loadAndFlushThreeTimes(rows, table, famStr);
LOG.debug("Taking snapshot and cloning table {}", table);
- admin.snapshot(test.getMethodName(), table);
- admin.cloneSnapshot(test.getMethodName(), clone);
+ admin.snapshot(TestMobUtils.getTableName(test), table);
+ admin.cloneSnapshot(TestMobUtils.getTableName(test), clone);
assertEquals("Should have 3 hlinks per region in MOB area from snapshot clone", 3 * numRegions,
getNumberOfMobFiles(clone, famStr));
loadAndFlushThreeTimes(rows, clone, famStr);
@@ -275,8 +295,11 @@ public class TestMobCompactionWithDefaults {
Thread.sleep(minAgeToArchive + 1000);
LOG.info("Cleaning up MOB files");
- // Cleanup again
- cleanerChore.cleanupObsoleteMobFiles(conf, table);
+
+ // run cleaner chore on each RS
+ for (ServerName sn : admin.getRegionServers()) {
+ HTU.getMiniHBaseCluster().getRegionServer(sn).getRSMobFileCleanerChore().chore();
+ }
assertEquals("After cleaning, we should have 1 MOB file per region based on size.", numRegions,
getNumberOfMobFiles(table, family));
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/TestMobFileCleanerChore.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/TestMobFileCleanerChore.java
index bac0408eb4a..6ce127b994a 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/TestMobFileCleanerChore.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/TestMobFileCleanerChore.java
@@ -18,6 +18,7 @@
package org.apache.hadoop.hbase.mob;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import java.io.IOException;
@@ -32,6 +33,8 @@ import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
+import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
+import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
import org.apache.hadoop.hbase.client.CompactionState;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
@@ -168,14 +171,40 @@ public class TestMobFileCleanerChore {
Thread.sleep(minAgeToArchive + 1000);
LOG.info("Cleaning up MOB files");
- // Cleanup again
+ // Cleanup
chore.cleanupObsoleteMobFiles(conf, table.getName());
+ // verify that nothing have happened
num = getNumberOfMobFiles(conf, table.getName(), new String(fam));
- assertEquals(1, num);
+ assertEquals(4, num);
long scanned = scanTable();
assertEquals(30, scanned);
+
+ // add a MOB file to with a name refering to a non-existing region
+ ColumnFamilyDescriptor familyDescriptor = ColumnFamilyDescriptorBuilder.newBuilder(fam)
+ .setMobEnabled(true).setMobThreshold(mobLen).setMaxVersions(1).build();
+ Path extraMOBFile = MobTestUtil.generateMOBFileForRegion(conf, table.getName(),
+ familyDescriptor, "nonExistentRegion");
+ num = getNumberOfMobFiles(conf, table.getName(), new String(fam));
+ assertEquals(5, num);
+
+ LOG.info("Waiting for {}ms", minAgeToArchive + 1000);
+
+ Thread.sleep(minAgeToArchive + 1000);
+ LOG.info("Cleaning up MOB files");
+ chore.cleanupObsoleteMobFiles(conf, table.getName());
+
+ // check that the extra file got deleted
+ num = getNumberOfMobFiles(conf, table.getName(), new String(fam));
+ assertEquals(4, num);
+
+ FileSystem fs = FileSystem.get(conf);
+ assertFalse(fs.exists(extraMOBFile));
+
+ scanned = scanTable();
+ assertEquals(30, scanned);
+
}
private long getNumberOfMobFiles(Configuration conf, TableName tableName, String family)
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/TestMobStoreCompaction.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/TestMobStoreCompaction.java
index 90b3d4085bf..123965c0eca 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/TestMobStoreCompaction.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/TestMobStoreCompaction.java
@@ -26,11 +26,14 @@ import static org.junit.Assert.assertTrue;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
+import java.util.UUID;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
@@ -65,6 +68,7 @@ import org.apache.hadoop.hbase.regionserver.InternalScanner;
import org.apache.hadoop.hbase.regionserver.RegionAsTable;
import org.apache.hadoop.hbase.regionserver.compactions.CompactionContext;
import org.apache.hadoop.hbase.regionserver.compactions.CompactionLifeCycleTracker;
+import org.apache.hadoop.hbase.regionserver.storefiletracker.StoreFileTrackerFactory;
import org.apache.hadoop.hbase.regionserver.throttle.NoLimitThroughputController;
import org.apache.hadoop.hbase.security.User;
import org.apache.hadoop.hbase.testclassification.MediumTests;
@@ -78,12 +82,15 @@ import org.junit.Rule;
import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.junit.rules.TestName;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Test mob store compaction
*/
+@RunWith(Parameterized.class)
@Category(MediumTests.class)
public class TestMobStoreCompaction {
@@ -108,13 +115,30 @@ public class TestMobStoreCompaction {
private final byte[] STARTROW = Bytes.toBytes(START_KEY);
private int compactionThreshold;
+ private Boolean useFileBasedSFT;
+
+ public TestMobStoreCompaction(Boolean useFileBasedSFT) {
+ this.useFileBasedSFT = useFileBasedSFT;
+ }
+
+ @Parameterized.Parameters
+ public static Collection<Boolean> data() {
+ Boolean[] data = { false, true };
+ return Arrays.asList(data);
+ }
+
private void init(Configuration conf, long mobThreshold) throws Exception {
+ if (useFileBasedSFT) {
+ conf.set(StoreFileTrackerFactory.TRACKER_IMPL,
+ "org.apache.hadoop.hbase.regionserver.storefiletracker.FileBasedStoreFileTracker");
+ }
+
this.conf = conf;
this.mobCellThreshold = mobThreshold;
HBaseTestingUtility UTIL = new HBaseTestingUtility(conf);
compactionThreshold = conf.getInt("hbase.hstore.compactionThreshold", 3);
- htd = UTIL.createTableDescriptor(name.getMethodName());
+ htd = UTIL.createTableDescriptor(TestMobUtils.getTableName(name));
hcd = new HColumnDescriptor(COLUMN_FAMILY);
hcd.setMobEnabled(true);
hcd.setMobThreshold(mobThreshold);
@@ -227,7 +251,7 @@ public class TestMobStoreCompaction {
Path basedir = new Path(hbaseRootDir, htd.getNameAsString());
List<Pair<byte[], String>> hfiles = new ArrayList<>(1);
for (int i = 0; i < compactionThreshold; i++) {
- Path hpath = new Path(basedir, "hfile" + i);
+ Path hpath = new Path(basedir, UUID.randomUUID().toString().replace("-", ""));
hfiles.add(Pair.newPair(COLUMN_FAMILY, hpath.toString()));
createHFile(hpath, i, dummyData);
}
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/TestMobUtils.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/TestMobUtils.java
index 247dfff77d1..32170e98b35 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/TestMobUtils.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/TestMobUtils.java
@@ -27,6 +27,7 @@ import org.apache.hadoop.hbase.util.Bytes;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.experimental.categories.Category;
+import org.junit.rules.TestName;
import org.apache.hbase.thirdparty.com.google.common.collect.ImmutableSet;
import org.apache.hbase.thirdparty.com.google.common.collect.ImmutableSetMultimap;
@@ -89,4 +90,8 @@ public class TestMobUtils {
assertTrue(testTable3Refs.contains("file3a"));
assertTrue(testTable3Refs.contains("file3b"));
}
+
+ public static String getTableName(TestName test) {
+ return test.getMethodName().replace("[", "-").replace("]", "");
+ }
}
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/TestMobFileCleanerChore.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/TestRSMobFileCleanerChore.java
similarity index 65%
copy from hbase-server/src/test/java/org/apache/hadoop/hbase/mob/TestMobFileCleanerChore.java
copy to hbase-server/src/test/java/org/apache/hadoop/hbase/mob/TestRSMobFileCleanerChore.java
index bac0408eb4a..98187631d96 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/TestMobFileCleanerChore.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/TestRSMobFileCleanerChore.java
@@ -22,21 +22,26 @@ import static org.junit.Assert.assertTrue;
import java.io.IOException;
import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
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.HBaseClassTestRule;
import org.apache.hadoop.hbase.HBaseTestingUtility;
-import org.apache.hadoop.hbase.HColumnDescriptor;
-import org.apache.hadoop.hbase.HTableDescriptor;
+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.ColumnFamilyDescriptor;
+import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
import org.apache.hadoop.hbase.client.CompactionState;
import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.RegionInfo;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Table;
+import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
import org.apache.hadoop.hbase.master.cleaner.TimeToLiveHFileCleaner;
import org.apache.hadoop.hbase.testclassification.MediumTests;
import org.apache.hadoop.hbase.util.Bytes;
@@ -50,17 +55,16 @@ import org.slf4j.LoggerFactory;
/**
* Mob file cleaner chore test. 1. Creates MOB table 2. Load MOB data and flushes it N times 3. Runs
- * major MOB compaction (N MOB files go to archive) 4. Verifies that number of MOB files in a mob
- * directory is N+1 5. Waits for a period of time larger than minimum age to archive 6. Runs Mob
- * cleaner chore 7 Verifies that number of MOB files in a mob directory is 1.
+ * major MOB compaction 4. Verifies that number of MOB files in a mob directory is N+1 5. Waits for
+ * a period of time larger than minimum age to archive 6. Runs Mob cleaner chore 7 Verifies that
+ * every old MOB file referenced from current RS was archived
*/
-@SuppressWarnings("deprecation")
@Category(MediumTests.class)
-public class TestMobFileCleanerChore {
- private static final Logger LOG = LoggerFactory.getLogger(TestMobFileCleanerChore.class);
+public class TestRSMobFileCleanerChore {
+ private static final Logger LOG = LoggerFactory.getLogger(TestRSMobFileCleanerChore.class);
@ClassRule
public static final HBaseClassTestRule CLASS_RULE =
- HBaseClassTestRule.forClass(TestMobFileCleanerChore.class);
+ HBaseClassTestRule.forClass(TestRSMobFileCleanerChore.class);
private HBaseTestingUtility HTU;
@@ -72,33 +76,30 @@ public class TestMobFileCleanerChore {
.toBytes("01234567890123456789012345678901234567890123456789012345678901234567890123456789");
private Configuration conf;
- private HTableDescriptor hdt;
- private HColumnDescriptor hcd;
+ private TableDescriptorBuilder.ModifyableTableDescriptor tableDescriptor;
+ private ColumnFamilyDescriptor familyDescriptor;
private Admin admin;
private Table table = null;
- private MobFileCleanerChore chore;
+ private RSMobFileCleanerChore chore;
private long minAgeToArchive = 10000;
- public TestMobFileCleanerChore() {
+ public TestRSMobFileCleanerChore() {
}
@Before
public void setUp() throws Exception {
HTU = new HBaseTestingUtility();
- hdt = HTU.createTableDescriptor("testMobCompactTable");
conf = HTU.getConfiguration();
initConf();
HTU.startMiniCluster();
admin = HTU.getAdmin();
- chore = new MobFileCleanerChore();
- hcd = new HColumnDescriptor(fam);
- hcd.setMobEnabled(true);
- hcd.setMobThreshold(mobLen);
- hcd.setMaxVersions(1);
- hdt.addFamily(hcd);
- table = HTU.createTable(hdt, null);
+ familyDescriptor = ColumnFamilyDescriptorBuilder.newBuilder(fam).setMobEnabled(true)
+ .setMobThreshold(mobLen).setMaxVersions(1).build();
+ tableDescriptor =
+ HTU.createModifyableTableDescriptor("testMobCompactTable").setColumnFamily(familyDescriptor);
+ table = HTU.createTable(tableDescriptor, Bytes.toByteArrays("1"));
}
private void initConf() {
@@ -140,42 +141,88 @@ public class TestMobFileCleanerChore {
@After
public void tearDown() throws Exception {
- admin.disableTable(hdt.getTableName());
- admin.deleteTable(hdt.getTableName());
+ admin.disableTable(tableDescriptor.getTableName());
+ admin.deleteTable(tableDescriptor.getTableName());
HTU.shutdownMiniCluster();
}
@Test
public void testMobFileCleanerChore() throws InterruptedException, IOException {
-
loadData(0, 10);
loadData(10, 10);
- loadData(20, 10);
+ // loadData(20, 10);
long num = getNumberOfMobFiles(conf, table.getName(), new String(fam));
- assertEquals(3, num);
+ assertEquals(2, num);
// Major compact
- admin.majorCompact(hdt.getTableName(), fam);
+ admin.majorCompact(tableDescriptor.getTableName(), fam);
// wait until compaction is complete
- while (admin.getCompactionState(hdt.getTableName()) != CompactionState.NONE) {
+ while (admin.getCompactionState(tableDescriptor.getTableName()) != CompactionState.NONE) {
Thread.sleep(100);
}
num = getNumberOfMobFiles(conf, table.getName(), new String(fam));
- assertEquals(4, num);
+ assertEquals(3, num);
// We have guarantee, that compcated file discharger will run during this pause
// because it has interval less than this wait time
LOG.info("Waiting for {}ms", minAgeToArchive + 1000);
Thread.sleep(minAgeToArchive + 1000);
LOG.info("Cleaning up MOB files");
- // Cleanup again
- chore.cleanupObsoleteMobFiles(conf, table.getName());
+
+ ServerName serverUsed = null;
+ List<RegionInfo> serverRegions = null;
+ for (ServerName sn : admin.getRegionServers()) {
+ serverRegions = admin.getRegions(sn);
+ if (serverRegions != null && serverRegions.size() > 0) {
+ // filtering out non test table regions
+ serverRegions = serverRegions.stream().filter(r -> r.getTable() == table.getName())
+ .collect(Collectors.toList());
+ // if such one is found use this rs
+ if (serverRegions.size() > 0) {
+ serverUsed = sn;
+ }
+ break;
+ }
+ }
+
+ chore = HTU.getMiniHBaseCluster().getRegionServer(serverUsed).getRSMobFileCleanerChore();
+
+ chore.chore();
num = getNumberOfMobFiles(conf, table.getName(), new String(fam));
- assertEquals(1, num);
+ assertEquals(3 - serverRegions.size(), num);
long scanned = scanTable();
- assertEquals(30, scanned);
+ assertEquals(20, scanned);
+
+ // creating a MOB file not referenced from the current RS
+ Path extraMOBFile = MobTestUtil.generateMOBFileForRegion(conf, table.getName(),
+ familyDescriptor, "nonExistentRegion");
+
+ // verifying the new MOBfile is added
+ num = getNumberOfMobFiles(conf, table.getName(), new String(fam));
+ assertEquals(4 - serverRegions.size(), num);
+
+ FileSystem fs = FileSystem.get(conf);
+ assertTrue(fs.exists(extraMOBFile));
+
+ LOG.info("Waiting for {}ms", minAgeToArchive + 1000);
+
+ Thread.sleep(minAgeToArchive + 1000);
+ LOG.info("Cleaning up MOB files");
+
+ // running chore again
+ chore.chore();
+
+ // the chore should only archive old MOB files that were referenced from the current RS
+ // the unrelated MOB file is still there
+ num = getNumberOfMobFiles(conf, table.getName(), new String(fam));
+ assertEquals(4 - serverRegions.size(), num);
+
+ assertTrue(fs.exists(extraMOBFile));
+
+ scanned = scanTable();
+ assertEquals(20, scanned);
}
private long getNumberOfMobFiles(Configuration conf, TableName tableName, String family)