diff --git a/hbase-endpoint/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationSyncUpToolWithBulkLoadedData.java b/hbase-endpoint/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationSyncUpToolWithBulkLoadedData.java index 278e6c5612..4f2af8d759 100644 --- a/hbase-endpoint/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationSyncUpToolWithBulkLoadedData.java +++ b/hbase-endpoint/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.coprocessor.CoprocessorHost; +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()); String classes = conf.get(CoprocessorHost.REGION_COPROCESSOR_CONF_KEY, ""); @@ -84,8 +86,8 @@ public class TestReplicationSyncUpToolWithBulkLoadedData extends TestReplication * Prepare 16 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); @@ -175,11 +177,17 @@ public class TestReplicationSyncUpToolWithBulkLoadedData extends TestReplication 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()), @@ -187,11 +195,18 @@ public class TestReplicationSyncUpToolWithBulkLoadedData extends TestReplication 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()), @@ -229,6 +244,28 @@ public class TestReplicationSyncUpToolWithBulkLoadedData extends TestReplication 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++) { 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 9b38ea27a0..7ceb89b30c 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.AtomicLong; 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; @@ -2382,7 +2383,7 @@ public class RSRpcServices implements HBaseRPCErrorHandler, filePaths.add(familyPath.getPath()); } // Check if the batch of files exceeds the current quota - enforcement.checkBulkLoad(regionServer.getFileSystem(), filePaths); + enforcement.checkBulkLoad(getFileSystem(filePaths), filePaths); } } @@ -3688,6 +3689,15 @@ public class RSRpcServices implements HBaseRPCErrorHandler, regionServer.executeProcedure(request.getProcId(), callable); } + 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()); + } + @Override @QosPriority(priority = HConstants.ADMIN_QOS) public ExecuteProceduresResponse executeProcedures(RpcController controller,