HBASE-22939 SpaceQuotas - Bulkload from different hdfs failed when space quotas are turned on. (apache#553)

Signed-off-by: Sakthi <sakthi@apache.org>
(cherry picked from commit f31301d)
Change-Id: Ic88e96cc1b83351424463fef52579a35a4074822

Author: Yiran-wu

hbase-endpoint/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationSyncUpToolWithBulkLoadedData.java

@@ -37,6 +37,7 @@
 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, "");
@@ -81,20 +83,21 @@ 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<String> randomHFileRangeListIterator = null;
-    Set<String> randomHFileRanges = new HashSet<>(16);
-    for (int i = 0; i < 16; i++) {
+    Set<String> randomHFileRanges = new HashSet<>(24);
+    for (int i = 0; i < 24; i++) {
       randomHFileRanges.add(UTIL1.getRandomUUID().toString());
     }
     List<String> randomHFileRangeList = new ArrayList<>(randomHFileRanges);
     Collections.sort(randomHFileRangeList);
     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);
 
@@ -175,23 +178,35 @@ private void loadAndReplicateHFiles(boolean verifyReplicationOnSlave,
     Iterator<String> 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()),
         Bytes.toBytes(randomHFileRangeListIterator.next()) } };
     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, 100);
+
     hfileRanges =
       new byte[][][] { new byte[][] { Bytes.toBytes(randomHFileRangeListIterator.next()),
         Bytes.toBytes(randomHFileRangeListIterator.next()) } };
-    loadAndValidateHFileReplication("HFileReplication_1", row, FAMILY, ht2Source, hfileRanges, 200);
+    loadFromOtherHDFSAndValidateHFileReplication("HFileReplication_1", row, FAMILY, ht2Source,
+      hfileRanges, 100);
 
     hfileRanges =
       new byte[][][] { new byte[][] { Bytes.toBytes(randomHFileRangeListIterator.next()),
@@ -229,6 +244,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++) {

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java

@@ -47,6 +47,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;
@@ -2376,7 +2377,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);
         }
       }
 
@@ -2481,6 +2482,15 @@ public CoprocessorServiceResponse execService(final RpcController controller,
     }
   }
 
+  private FileSystem getFileSystem(List<String> 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)