Skip to content

Commit

Permalink
HBASE-26598 Fix excessive connections in MajorCompactor (#3961)
Browse files Browse the repository at this point in the history 
Signed-off-by: Duo Zhang <zhangduo@apache.org>
  • Loading branch information
@msamirkhan @Apache9
msamirkhan authored and Apache9 committed Dec 23, 2021
1 parent 647184e commit 8e56696
Show file tree
Hide file tree
Showing 6 changed files with 57 additions and 77 deletions.
53 changes: 23 additions & 30 deletions ...-server/src/main/java/org/apache/hadoop/hbase/util/compaction/MajorCompactionRequest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -22,13 +22,11 @@
import java.util.List;
import java.util.Optional;
import java.util.Set;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.RegionInfo;
import org.apache.hadoop.hbase.regionserver.HRegionFileSystem;
import org.apache.hadoop.hbase.regionserver.StoreFileInfo;
Expand All @@ -45,26 +43,26 @@ class MajorCompactionRequest {

private static final Logger LOG = LoggerFactory.getLogger(MajorCompactionRequest.class);

protected final Configuration configuration;
protected final Connection connection;
protected final RegionInfo region;
private Set<String> stores;

MajorCompactionRequest(Configuration configuration, RegionInfo region) {
this.configuration = configuration;
MajorCompactionRequest(Connection connection, RegionInfo region) {
this.connection = connection;
this.region = region;
}

MajorCompactionRequest(Configuration configuration, RegionInfo region,
MajorCompactionRequest(Connection connection, RegionInfo region,
Set<String> stores) {
this(configuration, region);
this(connection, region);
this.stores = stores;
}

static Optional<MajorCompactionRequest> newRequest(Configuration configuration, RegionInfo info,
static Optional<MajorCompactionRequest> newRequest(Connection connection, RegionInfo info,
Set<String> stores, long timestamp) throws IOException {
MajorCompactionRequest request =
new MajorCompactionRequest(configuration, info, stores);
return request.createRequest(configuration, stores, timestamp);
new MajorCompactionRequest(connection, info, stores);
return request.createRequest(connection, stores, timestamp);
}

RegionInfo getRegion() {
Expand All @@ -79,28 +77,26 @@ void setStores(Set<String> stores) {
this.stores = stores;
}

Optional<MajorCompactionRequest> createRequest(Configuration configuration,
Optional<MajorCompactionRequest> createRequest(Connection connection,
Set<String> stores, long timestamp) throws IOException {
Set<String> familiesToCompact = getStoresRequiringCompaction(stores, timestamp);
MajorCompactionRequest request = null;
if (!familiesToCompact.isEmpty()) {
request = new MajorCompactionRequest(configuration, region, familiesToCompact);
request = new MajorCompactionRequest(connection, region, familiesToCompact);
}
return Optional.ofNullable(request);
}

Set<String> getStoresRequiringCompaction(Set<String> requestedStores, long timestamp)
throws IOException {
try(Connection connection = getConnection(configuration)) {
HRegionFileSystem fileSystem = getFileSystem(connection);
Set<String> familiesToCompact = Sets.newHashSet();
for (String family : requestedStores) {
if (shouldCFBeCompacted(fileSystem, family, timestamp)) {
familiesToCompact.add(family);
}
HRegionFileSystem fileSystem = getFileSystem();
Set<String> familiesToCompact = Sets.newHashSet();
for (String family : requestedStores) {
if (shouldCFBeCompacted(fileSystem, family, timestamp)) {
familiesToCompact.add(family);
}
return familiesToCompact;
}
return familiesToCompact;
}

boolean shouldCFBeCompacted(HRegionFileSystem fileSystem, String family, long ts)
Expand Down Expand Up @@ -142,10 +138,6 @@ protected boolean shouldIncludeStore(HRegionFileSystem fileSystem, String family
return false;
}

Connection getConnection(Configuration configuration) throws IOException {
return ConnectionFactory.createConnection(configuration);
}

protected boolean familyHasReferenceFile(HRegionFileSystem fileSystem, String family, long ts)
throws IOException {
List<Path> referenceFiles =
Expand All @@ -167,12 +159,13 @@ List<Path> getReferenceFilePaths(FileSystem fileSystem, Path familyDir)
return FSUtils.getReferenceFilePaths(fileSystem, familyDir);
}

HRegionFileSystem getFileSystem(Connection connection) throws IOException {
Admin admin = connection.getAdmin();
return HRegionFileSystem.openRegionFromFileSystem(admin.getConfiguration(),
CommonFSUtils.getCurrentFileSystem(admin.getConfiguration()), CommonFSUtils.getTableDir(
CommonFSUtils.getRootDir(admin.getConfiguration()), region.getTable()),
region, true);
HRegionFileSystem getFileSystem() throws IOException {
try (Admin admin = connection.getAdmin()) {
return HRegionFileSystem.openRegionFromFileSystem(admin.getConfiguration(),
CommonFSUtils.getCurrentFileSystem(admin.getConfiguration()),
CommonFSUtils.getTableDir(CommonFSUtils.getRootDir(admin.getConfiguration()),
region.getTable()), region, true);
}
}

@Override
Expand Down
33 changes: 15 additions & 18 deletions ...rver/src/main/java/org/apache/hadoop/hbase/util/compaction/MajorCompactionTTLRequest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,6 @@
import java.util.Collection;
import java.util.Map;
import java.util.Optional;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
import org.apache.hadoop.hbase.client.Connection;
Expand All @@ -44,40 +43,38 @@ public class MajorCompactionTTLRequest extends MajorCompactionRequest {

private static final Logger LOG = LoggerFactory.getLogger(MajorCompactionTTLRequest.class);

MajorCompactionTTLRequest(Configuration conf, RegionInfo region) {
super(conf, region);
MajorCompactionTTLRequest(Connection connection, RegionInfo region) {
super(connection, region);
}

static Optional<MajorCompactionRequest> newRequest(Configuration conf, RegionInfo info,
static Optional<MajorCompactionRequest> newRequest(Connection connection, RegionInfo info,
TableDescriptor htd) throws IOException {
MajorCompactionTTLRequest request = new MajorCompactionTTLRequest(conf, info);
return request.createRequest(conf, htd);
MajorCompactionTTLRequest request = new MajorCompactionTTLRequest(connection, info);
return request.createRequest(connection, htd);
}

private Optional<MajorCompactionRequest> createRequest(Configuration conf, TableDescriptor htd)
private Optional<MajorCompactionRequest> createRequest(Connection connection, TableDescriptor htd)
throws IOException {
Map<String, Long> familiesToCompact = getStoresRequiringCompaction(htd);
MajorCompactionRequest request = null;
if (!familiesToCompact.isEmpty()) {
LOG.debug("Compaction families for region: " + region + " CF: " + familiesToCompact.keySet());
request = new MajorCompactionTTLRequest(conf, region);
request = new MajorCompactionTTLRequest(connection, region);
}
return Optional.ofNullable(request);
}

Map<String, Long> getStoresRequiringCompaction(TableDescriptor htd) throws IOException {
try(Connection connection = getConnection(configuration)) {
HRegionFileSystem fileSystem = getFileSystem(connection);
Map<String, Long> familyTTLMap = Maps.newHashMap();
for (ColumnFamilyDescriptor descriptor : htd.getColumnFamilies()) {
long ts = getColFamilyCutoffTime(descriptor);
// If the table's TTL is forever, lets not compact any of the regions.
if (ts > 0 && shouldCFBeCompacted(fileSystem, descriptor.getNameAsString(), ts)) {
familyTTLMap.put(descriptor.getNameAsString(), ts);
}
HRegionFileSystem fileSystem = getFileSystem();
Map<String, Long> familyTTLMap = Maps.newHashMap();
for (ColumnFamilyDescriptor descriptor : htd.getColumnFamilies()) {
long ts = getColFamilyCutoffTime(descriptor);
// If the table's TTL is forever, lets not compact any of the regions.
if (ts > 0 && shouldCFBeCompacted(fileSystem, descriptor.getNameAsString(), ts)) {
familyTTLMap.put(descriptor.getNameAsString(), ts);
}
return familyTTLMap;
}
return familyTTLMap;
}

// If the CF has no TTL, return -1, else return the current time - TTL.
Expand Down
6 changes: 2 additions & 4 deletions hbase-server/src/main/java/org/apache/hadoop/hbase/util/compaction/MajorCompactor.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -201,8 +201,7 @@ void initializeWorkQueues() throws IOException {

protected Optional<MajorCompactionRequest> getMajorCompactionRequest(RegionInfo hri)
throws IOException {
return MajorCompactionRequest.newRequest(connection.getConfiguration(), hri, storesToCompact,
timestamp);
return MajorCompactionRequest.newRequest(connection, hri, storesToCompact, timestamp);
}

private Collection<ServerName> getServersToCompact(Set<ServerName> snSet) {
Expand Down Expand Up @@ -351,8 +350,7 @@ private void addNewRegions() {
for (HRegionLocation location : locations) {
if (location.getRegion().getRegionId() > timestamp) {
Optional<MajorCompactionRequest> compactionRequest = MajorCompactionRequest
.newRequest(connection.getConfiguration(), location.getRegion(), storesToCompact,
timestamp);
.newRequest(connection, location.getRegion(), storesToCompact, timestamp);
compactionRequest.ifPresent(request -> clusterCompactionQueues
.addToCompactionQueue(location.getServerName(), request));
}
Expand Down
4 changes: 2 additions & 2 deletions hbase-server/src/main/java/org/apache/hadoop/hbase/util/compaction/MajorCompactorTTL.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -76,7 +76,7 @@ protected MajorCompactorTTL() {
@Override
protected Optional<MajorCompactionRequest> getMajorCompactionRequest(RegionInfo hri)
throws IOException {
return MajorCompactionTTLRequest.newRequest(connection.getConfiguration(), hri, htd);
return MajorCompactionTTLRequest.newRequest(connection, hri, htd);
}

@Override
Expand Down Expand Up @@ -171,4 +171,4 @@ public int run(String[] args) throws Exception {
public static void main(String[] args) throws Exception {
ToolRunner.run(HBaseConfiguration.create(), new MajorCompactorTTL(), args);
}
}
}
20 changes: 8 additions & 12 deletions ...ver/src/test/java/org/apache/hadoop/hbase/util/compaction/TestMajorCompactionRequest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,6 @@
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.ArgumentMatchers.isA;
import static org.mockito.Matchers.anyString;
import static org.mockito.Mockito.doReturn;
Expand All @@ -36,7 +35,6 @@
import java.util.stream.Collectors;

import org.apache.commons.lang3.RandomStringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
Expand Down Expand Up @@ -81,13 +79,13 @@ public class TestMajorCompactionRequest {
List<StoreFileInfo> storeFiles = mockStoreFiles(regionStoreDir, 5, 10);
MajorCompactionRequest request = makeMockRequest(storeFiles, false);
Optional<MajorCompactionRequest> result =
request.createRequest(mock(Configuration.class), Sets.newHashSet(FAMILY), 100);
request.createRequest(mock(Connection.class), Sets.newHashSet(FAMILY), 100);
assertTrue(result.isPresent());

// store files newer than timestamp
storeFiles = mockStoreFiles(regionStoreDir, 5, 101);
request = makeMockRequest(storeFiles, false);
result = request.createRequest(mock(Configuration.class), Sets.newHashSet(FAMILY), 100);
result = request.createRequest(mock(Connection.class), Sets.newHashSet(FAMILY), 100);
assertFalse(result.isPresent());
}

Expand All @@ -100,19 +98,18 @@ public class TestMajorCompactionRequest {
HRegion region =
HBaseTestingUtility.createRegionAndWAL(hri, rootRegionDir, UTILITY.getConfiguration(), htd);

Configuration configuration = mock(Configuration.class);
Connection connection = mock(Connection.class);
// the reference file timestamp is newer
List<StoreFileInfo> storeFiles = mockStoreFiles(regionStoreDir, 4, 101);
List<Path> paths = storeFiles.stream().map(StoreFileInfo::getPath).collect(Collectors.toList());
// the files that are referenced are older, thus we still compact.
HRegionFileSystem fileSystem =
mockFileSystem(region.getRegionInfo(), true, storeFiles, 50);
MajorCompactionRequest majorCompactionRequest = spy(new MajorCompactionRequest(configuration,
MajorCompactionRequest majorCompactionRequest = spy(new MajorCompactionRequest(connection,
region.getRegionInfo(), Sets.newHashSet(FAMILY)));
doReturn(mock(Connection.class)).when(majorCompactionRequest).getConnection(eq(configuration));
doReturn(paths).when(majorCompactionRequest).getReferenceFilePaths(any(FileSystem.class),
any(Path.class));
doReturn(fileSystem).when(majorCompactionRequest).getFileSystem(any(Connection.class));
doReturn(fileSystem).when(majorCompactionRequest).getFileSystem();
Set<String> result =
majorCompactionRequest.getStoresRequiringCompaction(Sets.newHashSet("a"), 100);
assertEquals(FAMILY, Iterables.getOnlyElement(result));
Expand Down Expand Up @@ -158,16 +155,15 @@ protected List<StoreFileInfo> mockStoreFiles(Path regionStoreDir, int howMany, l

private MajorCompactionRequest makeMockRequest(List<StoreFileInfo> storeFiles,
boolean references) throws IOException {
Configuration configuration = mock(Configuration.class);
Connection connection = mock(Connection.class);
RegionInfo regionInfo = mock(RegionInfo.class);
when(regionInfo.getEncodedName()).thenReturn("HBase");
when(regionInfo.getTable()).thenReturn(TableName.valueOf("foo"));
MajorCompactionRequest request =
new MajorCompactionRequest(configuration, regionInfo, Sets.newHashSet("a"));
new MajorCompactionRequest(connection, regionInfo, Sets.newHashSet("a"));
MajorCompactionRequest spy = spy(request);
HRegionFileSystem fileSystem = mockFileSystem(regionInfo, references, storeFiles);
doReturn(fileSystem).when(spy).getFileSystem(isA(Connection.class));
doReturn(mock(Connection.class)).when(spy).getConnection(eq(configuration));
doReturn(fileSystem).when(spy).getFileSystem();
return spy;
}
}
18 changes: 7 additions & 11 deletions .../src/test/java/org/apache/hadoop/hbase/util/compaction/TestMajorCompactionTTLRequest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -20,8 +20,6 @@

import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.mockito.Matchers.eq;
import static org.mockito.Matchers.isA;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
Expand All @@ -31,7 +29,6 @@
import java.util.List;
import java.util.Optional;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseClassTestRule;
import org.apache.hadoop.hbase.TableName;
Expand Down Expand Up @@ -72,29 +69,28 @@ public void testStoresNeedingCompaction() throws Exception {
MajorCompactionTTLRequest request = makeMockRequest(storeFiles);
// All files are <= 100, so region should not be compacted.
Optional<MajorCompactionRequest> result =
request.createRequest(mock(Configuration.class), Sets.newHashSet(FAMILY), 10);
request.createRequest(mock(Connection.class), Sets.newHashSet(FAMILY), 10);
assertFalse(result.isPresent());

// All files are <= 100, so region should not be compacted yet.
result = request.createRequest(mock(Configuration.class), Sets.newHashSet(FAMILY), 100);
result = request.createRequest(mock(Connection.class), Sets.newHashSet(FAMILY), 100);
assertFalse(result.isPresent());

// All files are <= 100, so they should be considered for compaction
result = request.createRequest(mock(Configuration.class), Sets.newHashSet(FAMILY), 101);
result = request.createRequest(mock(Connection.class), Sets.newHashSet(FAMILY), 101);
assertTrue(result.isPresent());
}

private MajorCompactionTTLRequest makeMockRequest(List<StoreFileInfo> storeFiles)
throws IOException {
Configuration configuration = mock(Configuration.class);
Connection connection = mock(Connection.class);
RegionInfo regionInfo = mock(RegionInfo.class);
when(regionInfo.getEncodedName()).thenReturn("HBase");
when(regionInfo.getTable()).thenReturn(TableName.valueOf("foo"));
MajorCompactionTTLRequest request = new MajorCompactionTTLRequest(configuration, regionInfo);
MajorCompactionTTLRequest request = new MajorCompactionTTLRequest(connection, regionInfo);
MajorCompactionTTLRequest spy = spy(request);
HRegionFileSystem fileSystem = mockFileSystem(regionInfo, false, storeFiles);
doReturn(fileSystem).when(spy).getFileSystem(isA(Connection.class));
doReturn(mock(Connection.class)).when(spy).getConnection(eq(configuration));
doReturn(fileSystem).when(spy).getFileSystem();
return spy;
}
}
}

0 comments on commit 8e56696

Please sign in to comment.