From febd226a8e718e33f4778728af0656a6a6131f2c Mon Sep 17 00:00:00 2001 From: Yiran Wu Date: Fri, 30 Aug 2019 15:54:38 +0800 Subject: [PATCH] HBASE-22939 SpaceQuotas - Bulkload from different hdfs failed when space quotas are turned on. --- .../hbase/regionserver/RSRpcServices.java | 12 ++++- ...plicationSyncUpToolWithBulkLoadedData.java | 53 +++++++++++++++---- 2 files changed, 55 insertions(+), 10 deletions(-) diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java index 61354fda5888..fe1c43a3343d 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java @@ -45,6 +45,7 @@ import java.util.concurrent.atomic.LongAdder; import org.apache.commons.lang3.mutable.MutableObject; import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.hbase.ByteBufferExtendedCell; import org.apache.hadoop.hbase.CacheEvictionStats; @@ -2404,7 +2405,7 @@ public BulkLoadHFileResponse bulkLoadHFile(final RpcController controller, filePaths.add(familyPath.getPath()); } // Check if the batch of files exceeds the current quota - sizeToBeLoaded = enforcement.computeBulkLoadSize(regionServer.getFileSystem(), filePaths); + sizeToBeLoaded = enforcement.computeBulkLoadSize(getFileSystem(filePaths), filePaths); } } // secure bulk load @@ -2492,6 +2493,15 @@ public CoprocessorServiceResponse execService(final RpcController controller, } } + private FileSystem getFileSystem(List filePaths) throws IOException { + if (filePaths.isEmpty()) { + // local hdfs + return regionServer.getFileSystem(); + } + // source hdfs + return new Path(filePaths.get(0)).getFileSystem(regionServer.getConfiguration()); + } + private com.google.protobuf.Message execServiceOnRegion(HRegion region, final ClientProtos.CoprocessorServiceCall serviceCall) throws IOException { // ignore the passed in controller (from the serialized call) diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationSyncUpToolWithBulkLoadedData.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationSyncUpToolWithBulkLoadedData.java index 5f10ef96285a..6247c22ff256 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationSyncUpToolWithBulkLoadedData.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationSyncUpToolWithBulkLoadedData.java @@ -37,6 +37,7 @@ import org.apache.hadoop.hbase.HConstants; import org.apache.hadoop.hbase.TableName; import org.apache.hadoop.hbase.client.Table; +import org.apache.hadoop.hbase.quotas.QuotaUtil; import org.apache.hadoop.hbase.replication.regionserver.TestSourceFSConfigurationProvider; import org.apache.hadoop.hbase.testclassification.LargeTests; import org.apache.hadoop.hbase.testclassification.ReplicationTests; @@ -63,6 +64,7 @@ public class TestReplicationSyncUpToolWithBulkLoadedData extends TestReplication protected void customizeClusterConf(Configuration conf) { conf.setBoolean(HConstants.REPLICATION_BULKLOAD_ENABLE_KEY, true); conf.set(HConstants.REPLICATION_CLUSTER_ID, "12345"); + conf.setBoolean(QuotaUtil.QUOTA_CONF_KEY, true); conf.set("hbase.replication.source.fs.conf.provider", TestSourceFSConfigurationProvider.class.getCanonicalName()); } @@ -76,11 +78,11 @@ public void testSyncUpTool() throws Exception { setupReplication(); /** - * Prepare 16 random hfile ranges required for creating hfiles + * Prepare 24 random hfile ranges required for creating hfiles */ Iterator randomHFileRangeListIterator = null; - Set randomHFileRanges = new HashSet<>(16); - for (int i = 0; i < 16; i++) { + Set randomHFileRanges = new HashSet<>(24); + for (int i = 0; i < 24; i++) { randomHFileRanges.add(UTIL1.getRandomUUID().toString()); } List randomHFileRangeList = new ArrayList<>(randomHFileRanges); @@ -88,8 +90,9 @@ public void testSyncUpTool() throws Exception { randomHFileRangeListIterator = randomHFileRangeList.iterator(); /** - * at Master: t1_syncup: Load 100 rows into cf1, and 3 rows into norep t2_syncup: Load 200 rows - * into cf1, and 3 rows into norep verify correctly replicated to slave + * at Master: t1_syncup: Load 50 rows into cf1, and 50 rows from other hdfs into cf1, and 3 + * rows into norep t2_syncup: Load 100 rows into cf1, and 100 rows from other hdfs into cf1, + * and 3 rows into norep verify correctly replicated to slave */ loadAndReplicateHFiles(true, randomHFileRangeListIterator); @@ -170,11 +173,17 @@ private void loadAndReplicateHFiles(boolean verifyReplicationOnSlave, Iterator randomHFileRangeListIterator) throws Exception { LOG.debug("loadAndReplicateHFiles"); - // Load 100 + 3 hfiles to t1_syncup. + // Load 50 + 50 + 3 hfiles to t1_syncup. byte[][][] hfileRanges = new byte[][][] { new byte[][] { Bytes.toBytes(randomHFileRangeListIterator.next()), Bytes.toBytes(randomHFileRangeListIterator.next()) } }; - loadAndValidateHFileReplication("HFileReplication_1", row, FAMILY, ht1Source, hfileRanges, 100); + loadAndValidateHFileReplication("HFileReplication_1", row, FAMILY, ht1Source, hfileRanges, 50); + + hfileRanges = + new byte[][][] { new byte[][] { Bytes.toBytes(randomHFileRangeListIterator.next()), + Bytes.toBytes(randomHFileRangeListIterator.next()) } }; + loadFromOtherHDFSAndValidateHFileReplication("HFileReplication_1", row, FAMILY, ht1Source, + hfileRanges, 50); hfileRanges = new byte[][][] { new byte[][] { Bytes.toBytes(randomHFileRangeListIterator.next()), @@ -182,11 +191,17 @@ private void loadAndReplicateHFiles(boolean verifyReplicationOnSlave, loadAndValidateHFileReplication("HFileReplication_1", row, NO_REP_FAMILY, ht1Source, hfileRanges, 3); - // Load 200 + 3 hfiles to t2_syncup. + // Load 100 + 100 + 3 hfiles to t2_syncup. hfileRanges = new byte[][][] { new byte[][] { Bytes.toBytes(randomHFileRangeListIterator.next()), Bytes.toBytes(randomHFileRangeListIterator.next()) } }; - loadAndValidateHFileReplication("HFileReplication_1", row, FAMILY, ht2Source, hfileRanges, 200); + loadAndValidateHFileReplication("HFileReplication_1", row, FAMILY, ht2Source, hfileRanges, 100); + + hfileRanges = + new byte[][][] { new byte[][] { Bytes.toBytes(randomHFileRangeListIterator.next()), + Bytes.toBytes(randomHFileRangeListIterator.next()) } }; + loadFromOtherHDFSAndValidateHFileReplication("HFileReplication_1", row, FAMILY, ht2Source, + hfileRanges, 100); hfileRanges = new byte[][][] { new byte[][] { Bytes.toBytes(randomHFileRangeListIterator.next()), @@ -224,6 +239,26 @@ private void loadAndValidateHFileReplication(String testName, byte[] row, byte[] loader.bulkLoad(tableName, dir); } + private void loadFromOtherHDFSAndValidateHFileReplication(String testName, byte[] row, byte[] fam, + Table source, byte[][][] hfileRanges, int numOfRows) throws Exception { + Path dir = UTIL2.getDataTestDirOnTestFS(testName); + FileSystem fs = UTIL2.getTestFileSystem(); + dir = dir.makeQualified(fs); + Path familyDir = new Path(dir, Bytes.toString(fam)); + + int hfileIdx = 0; + for (byte[][] range : hfileRanges) { + byte[] from = range[0]; + byte[] to = range[1]; + HFileTestUtil.createHFile(UTIL2.getConfiguration(), fs, + new Path(familyDir, "hfile_" + hfileIdx++), fam, row, from, to, numOfRows); + } + + final TableName tableName = source.getName(); + BulkLoadHFiles loader = BulkLoadHFiles.create(UTIL1.getConfiguration()); + loader.bulkLoad(tableName, dir); + } + private void wait(Table target, int expectedCount, String msg) throws IOException, InterruptedException { for (int i = 0; i < NB_RETRIES; i++) {