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:16 UTC
[hbase] branch branch-2 updated (1986a6ec313 -> d073404706b)
This is an automated email from the ASF dual-hosted git repository.
psomogyi pushed a change to branch branch-2
in repository https://gitbox.apache.org/repos/asf/hbase.git
from 1986a6ec313 HBASE-24163 MOB compactor implementations should use format specifiers when calling String.format
new 1176b484d13 HBASE-25970 MOB data loss - incorrect concatenation of MOB_FILE_REFS (#3355)
new d073404706b HBASE-26969:Eliminate MOB renames when SFT is enabled
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../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 | 43 +++-
.../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 | 97 ++++++++
...erChore.java => TestRSMobFileCleanerChore.java} | 113 ++++++---
22 files changed, 790 insertions(+), 439 deletions(-)
create mode 100644 hbase-server/src/main/java/org/apache/hadoop/hbase/mob/RSMobFileCleanerChore.java
delete mode 100644 hbase-server/src/test/java/org/apache/hadoop/hbase/mob/TestMobCompactionBase.java
delete mode 100644 hbase-server/src/test/java/org/apache/hadoop/hbase/mob/TestMobCompactionRegularMode.java
create mode 100644 hbase-server/src/test/java/org/apache/hadoop/hbase/mob/TestMobUtils.java
copy hbase-server/src/test/java/org/apache/hadoop/hbase/mob/{TestMobFileCleanerChore.java => TestRSMobFileCleanerChore.java} (65%)
[hbase] 01/02: HBASE-25970 MOB data loss - incorrect concatenation of MOB_FILE_REFS (#3355)
Posted by ps...@apache.org.
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 1176b484d13dd8ed83cfd397b8b170e8ab534786
Author: Peter Somogyi <ps...@apache.org>
AuthorDate: Sat Jun 5 08:57:26 2021 +0200
HBASE-25970 MOB data loss - incorrect concatenation of MOB_FILE_REFS (#3355)
Signed-off-by: Wellington Chevreuil <wc...@apache.org>
Signed-off-by: Pankaj Kumar<pa...@apache.org>
---
.../java/org/apache/hadoop/hbase/mob/MobUtils.java | 12 +--
.../org/apache/hadoop/hbase/mob/TestMobUtils.java | 92 ++++++++++++++++++++++
2 files changed, 98 insertions(+), 6 deletions(-)
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 f408c9642a7..a409c9f42c4 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
@@ -737,20 +737,20 @@ public final class MobUtils {
StringBuilder sb = new StringBuilder(100 + mobRefSet.size() * 105);
boolean doubleSlash = false;
for (TableName tableName : mobRefSet.keySet()) {
+ if (doubleSlash) {
+ sb.append("//");
+ } else {
+ doubleSlash = true;
+ }
sb.append(tableName).append("/");
boolean comma = false;
for (String refs : mobRefSet.get(tableName)) {
- sb.append(refs);
if (comma) {
sb.append(",");
} else {
comma = true;
}
- }
- if (doubleSlash) {
- sb.append("//");
- } else {
- doubleSlash = true;
+ sb.append(refs);
}
}
return Bytes.toBytes(sb.toString());
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
new file mode 100644
index 00000000000..247dfff77d1
--- /dev/null
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/TestMobUtils.java
@@ -0,0 +1,92 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.mob;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import org.apache.hadoop.hbase.HBaseClassTestRule;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.testclassification.SmallTests;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import org.apache.hbase.thirdparty.com.google.common.collect.ImmutableSet;
+import org.apache.hbase.thirdparty.com.google.common.collect.ImmutableSetMultimap;
+
+@Category(SmallTests.class)
+public class TestMobUtils {
+
+ @ClassRule
+ public static final HBaseClassTestRule CLASS_RULE =
+ HBaseClassTestRule.forClass(TestMobUtils.class);
+ public static final TableName TEST_TABLE_1 = TableName.valueOf("testTable1");
+ public static final TableName TEST_TABLE_2 = TableName.valueOf("testTable2");
+ public static final TableName TEST_TABLE_3 = TableName.valueOf("testTable3");
+
+ @Test
+ public void serializeSingleMobFileRefs() {
+ ImmutableSetMultimap<TableName, String> mobRefSet =
+ ImmutableSetMultimap.<TableName, String> builder().putAll(TEST_TABLE_1, "file1a").build();
+ byte[] result = MobUtils.serializeMobFileRefs(mobRefSet);
+ assertEquals("testTable1/file1a", Bytes.toString(result));
+ }
+
+ @Test
+ public void serializeMultipleMobFileRefs() {
+ ImmutableSetMultimap<TableName, String> mobRefSet =
+ ImmutableSetMultimap.<TableName, String> builder().putAll(TEST_TABLE_1, "file1a", "file1b")
+ .putAll(TEST_TABLE_2, "file2a").putAll(TEST_TABLE_3, "file3a", "file3b").build();
+ byte[] result = MobUtils.serializeMobFileRefs(mobRefSet);
+ assertEquals("testTable1/file1a,file1b//testTable2/file2a//testTable3/file3a,file3b",
+ Bytes.toString(result));
+ }
+
+ @Test
+ public void deserializeSingleMobFileRefs() {
+ ImmutableSetMultimap<TableName, String> mobRefSet =
+ MobUtils.deserializeMobFileRefs(Bytes.toBytes("testTable1/file1a")).build();
+ assertEquals(1, mobRefSet.size());
+ ImmutableSet<String> testTable1Refs = mobRefSet.get(TEST_TABLE_1);
+ assertEquals(1, testTable1Refs.size());
+ assertTrue(testTable1Refs.contains("file1a"));
+ }
+
+ @Test
+ public void deserializeMultipleMobFileRefs() {
+ ImmutableSetMultimap<TableName,
+ String> mobRefSet = MobUtils
+ .deserializeMobFileRefs(
+ Bytes.toBytes("testTable1/file1a,file1b//testTable2/file2a//testTable3/file3a,file3b"))
+ .build();
+ assertEquals(5, mobRefSet.size());
+ ImmutableSet<String> testTable1Refs = mobRefSet.get(TEST_TABLE_1);
+ ImmutableSet<String> testTable2Refs = mobRefSet.get(TEST_TABLE_2);
+ ImmutableSet<String> testTable3Refs = mobRefSet.get(TEST_TABLE_3);
+ assertEquals(2, testTable1Refs.size());
+ assertEquals(1, testTable2Refs.size());
+ assertEquals(2, testTable3Refs.size());
+ assertTrue(testTable1Refs.contains("file1a"));
+ assertTrue(testTable1Refs.contains("file1b"));
+ assertTrue(testTable2Refs.contains("file2a"));
+ assertTrue(testTable3Refs.contains("file3a"));
+ assertTrue(testTable3Refs.contains("file3b"));
+ }
+}
[hbase] 02/02: HBASE-26969:Eliminate MOB renames when SFT is enabled
Posted by ps...@apache.org.
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)