HBASE-30238 Add bandwidth throttling for bulkload HFile copy during replication

Introduce hbase.replication.bulkload.copy.bandwidth.mb to rate-limit HFile copy from source HDFS in HFileReplicator.Copier.

The limit is enforced through a shared RateLimiter across copy threads, and ReplicationSink updates the rate through configuration reload without requiring RegionServer restart.

Author: juke-mini666

hbase-server/src/main/java/org/apache/hadoop/hbase/replication/ReplicationSinkServiceImpl.java

@@ -28,6 +28,7 @@
 import org.apache.hadoop.hbase.ScheduledChore;
 import org.apache.hadoop.hbase.Server;
 import org.apache.hadoop.hbase.Stoppable;
+import org.apache.hadoop.hbase.conf.ConfigurationObserver;
 import org.apache.hadoop.hbase.regionserver.HRegionServer;
 import org.apache.hadoop.hbase.regionserver.RegionServerCoprocessorHost;
 import org.apache.hadoop.hbase.regionserver.ReplicationSinkService;
@@ -41,7 +42,7 @@
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos;
 
 @InterfaceAudience.Private
-public class ReplicationSinkServiceImpl implements ReplicationSinkService {
+public class ReplicationSinkServiceImpl implements ReplicationSinkService, ConfigurationObserver {
   private static final Logger LOG = LoggerFactory.getLogger(ReplicationSinkServiceImpl.class);
 
   private Configuration conf;
@@ -90,6 +91,14 @@ public void stopReplicationService() {
     }
   }
 
+  @Override
+  public void onConfigurationChange(Configuration conf) {
+    this.conf = conf;
+    if (this.replicationSink != null) {
+      this.replicationSink.onConfigurationChange(conf);
+    }
+  }
+
   @Override
   public ReplicationLoad refreshAndGetReplicationLoad() {
     if (replicationLoad == null) {

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

@@ -20,7 +20,9 @@
 import java.io.Closeable;
 import java.io.FileNotFoundException;
 import java.io.IOException;
+import java.io.InputStream;
 import java.io.InterruptedIOException;
+import java.io.OutputStream;
 import java.math.BigInteger;
 import java.util.ArrayList;
 import java.util.Deque;
@@ -37,7 +39,6 @@
 import java.util.concurrent.TimeUnit;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.FileUtil;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.hbase.HConstants;
@@ -57,6 +58,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.apache.hbase.thirdparty.com.google.common.util.concurrent.RateLimiter;
 import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
 
 /**
@@ -75,10 +77,20 @@ public class HFileReplicator implements Closeable {
   public static final String REPLICATION_BULKLOAD_COPY_HFILES_PERTHREAD_KEY =
     "hbase.replication.bulkload.copy.hfiles.perthread";
   public static final int REPLICATION_BULKLOAD_COPY_HFILES_PERTHREAD_DEFAULT = 10;
+  /**
+   * Bandwidth limit in MB/s for copying HFiles from source cluster during bulkload replication. 0
+   * means no limit. Can be changed dynamically via configuration reload.
+   */
+  public static final String REPLICATION_BULKLOAD_COPY_BANDWIDTH_MB_KEY =
+    "hbase.replication.bulkload.copy.bandwidth.mb";
+  public static final double REPLICATION_BULKLOAD_COPY_BANDWIDTH_MB_DEFAULT = 0;
 
   private static final Logger LOG = LoggerFactory.getLogger(HFileReplicator.class);
   private static final String UNDERSCORE = "_";
   private final static FsPermission PERM_ALL_ACCESS = FsPermission.valueOf("-rwxrwxrwx");
+  private static final int COPY_BUFFER_SIZE = 65536;
+  // null means no throttling
+  private volatile RateLimiter rateLimiter;
 
   private Configuration sourceClusterConf;
   private String sourceBaseNamespaceDirPath;
@@ -99,6 +111,14 @@ public HFileReplicator(Configuration sourceClusterConf, String sourceBaseNamespa
     String sourceHFileArchiveDirPath, Map<String, List<Pair<byte[], List<String>>>> tableQueueMap,
     Configuration conf, AsyncClusterConnection connection, List<String> sourceClusterIds)
     throws IOException {
+    this(sourceClusterConf, sourceBaseNamespaceDirPath, sourceHFileArchiveDirPath, tableQueueMap,
+      conf, connection, sourceClusterIds, RateLimiter.create(Double.MAX_VALUE));
+  }
+
+  public HFileReplicator(Configuration sourceClusterConf, String sourceBaseNamespaceDirPath,
+    String sourceHFileArchiveDirPath, Map<String, List<Pair<byte[], List<String>>>> tableQueueMap,
+    Configuration conf, AsyncClusterConnection connection, List<String> sourceClusterIds,
+    RateLimiter rateLimiter) throws IOException {
     this.sourceClusterConf = sourceClusterConf;
     this.sourceBaseNamespaceDirPath = sourceBaseNamespaceDirPath;
     this.sourceHFileArchiveDirPath = sourceHFileArchiveDirPath;
@@ -120,6 +140,7 @@ public HFileReplicator(Configuration sourceClusterConf, String sourceBaseNamespa
       REPLICATION_BULKLOAD_COPY_HFILES_PERTHREAD_DEFAULT);
 
     sinkFs = FileSystem.get(conf);
+    this.rateLimiter = rateLimiter;
   }
 
   @Override
@@ -336,16 +357,13 @@ public Void call() throws IOException {
         sourceHFilePath = new Path(sourceBaseNamespaceDirPath, hfiles.get(i));
         localHFilePath = new Path(stagingDir, sourceHFilePath.getName());
         try {
-          FileUtil.copy(sourceFs, sourceHFilePath, sinkFs, localHFilePath, false, conf);
-          // If any other exception other than FNFE then we will fail the replication requests and
-          // source will retry to replicate these data.
+          copyWithThrottle(sourceHFilePath, localHFilePath);
         } catch (FileNotFoundException e) {
           LOG.info("Failed to copy hfile from " + sourceHFilePath + " to " + localHFilePath
             + ". Trying to copy from hfile archive directory.", e);
           sourceHFilePath = new Path(sourceHFileArchiveDirPath, hfiles.get(i));
-
           try {
-            FileUtil.copy(sourceFs, sourceHFilePath, sinkFs, localHFilePath, false, conf);
+            copyWithThrottle(sourceHFilePath, localHFilePath);
           } catch (FileNotFoundException e1) {
             // This will mean that the hfile does not exists any where in source cluster FS. So we
             // cannot do anything here just log and continue.
@@ -358,5 +376,16 @@ public Void call() throws IOException {
       }
       return null;
     }
+
+    private void copyWithThrottle(Path src, Path dst) throws IOException {
+      try (InputStream in = sourceFs.open(src); OutputStream out = sinkFs.create(dst)) {
+        byte[] buf = new byte[COPY_BUFFER_SIZE];
+        int bytesRead;
+        while ((bytesRead = in.read(buf)) >= 0) {
+          rateLimiter.acquire(bytesRead);
+          out.write(buf, 0, bytesRead);
+        }
+      }
+    }
   }
 }

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

@@ -42,6 +42,7 @@
 import java.util.stream.Collectors;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.conf.ConfigurationObserver;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hbase.Cell;
 import org.apache.hadoop.hbase.CellUtil;
@@ -71,6 +72,7 @@
 import org.slf4j.LoggerFactory;
 
 import org.apache.hbase.thirdparty.com.google.common.collect.Lists;
+import org.apache.hbase.thirdparty.com.google.common.util.concurrent.RateLimiter;
 
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.WALEntry;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos;
@@ -94,7 +96,7 @@
  * TODO make this class more like ReplicationSource wrt log handling
  */
 @InterfaceAudience.Private
-public class ReplicationSink {
+public class ReplicationSink implements ConfigurationObserver {
 
   private static final Logger LOG = LoggerFactory.getLogger(ReplicationSink.class);
   private final Configuration conf;
@@ -108,6 +110,7 @@ public class ReplicationSink {
   private long hfilesReplicated = 0;
   private SourceFSConfigurationProvider provider;
   private WALEntrySinkFilter walEntrySinkFilter;
+  private final RateLimiter bulkLoadCopyRateLimiter = RateLimiter.create(Double.MAX_VALUE);
 
   /**
    * Row size threshold for multi requests above which a warning is logged
@@ -143,6 +146,25 @@ public ReplicationSink(Configuration conf, RegionServerCoprocessorHost rsServerH
       throw new IllegalArgumentException(
         "Configured source fs configuration provider class " + className + " throws error.", e);
     }
+    updateBulkLoadCopyBandwidth(conf);
+  }
+
+  @Override
+  public void onConfigurationChange(Configuration newConf) {
+    updateBulkLoadCopyBandwidth(newConf);
+  }
+
+  double getBulkLoadCopyRateLimiterRate() {
+    return bulkLoadCopyRateLimiter.getRate();
+  }
+
+  private void updateBulkLoadCopyBandwidth(Configuration conf) {
+    double bandwidthMb = conf.getDouble(HFileReplicator.REPLICATION_BULKLOAD_COPY_BANDWIDTH_MB_KEY,
+      HFileReplicator.REPLICATION_BULKLOAD_COPY_BANDWIDTH_MB_DEFAULT);
+    double newRate = bandwidthMb <= 0 ? Double.MAX_VALUE : bandwidthMb * 1024 * 1024;
+    bulkLoadCopyRateLimiter.setRate(newRate);
+    LOG.info("Bulkload copy bandwidth updated: {}",
+      bandwidthMb <= 0 ? "unlimited" : bandwidthMb + " MB/s");
   }
 
   private WALEntrySinkFilter setupWALEntrySinkFilter() throws IOException {
@@ -319,7 +341,7 @@ public void replicateEntries(List<WALEntry> entries, final ExtendedCellScanner c
             Configuration providerConf = this.provider.getConf(this.conf, replicationClusterId);
             try (HFileReplicator hFileReplicator = new HFileReplicator(providerConf,
               sourceBaseNamespaceDirPath, sourceHFileArchiveDirPath, bulkLoadHFileMap, conf,
-              getConnection(), entry.getKey())) {
+              getConnection(), entry.getKey(), bulkLoadCopyRateLimiter)) {
               hFileReplicator.replicate();
               LOG.debug("Finished replicating {} bulk loaded data", entry.getKey().toString());
             }

hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestHFileReplicatorBandwidth.java

@@ -0,0 +1,138 @@
+/*
+ * 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.replication.regionserver;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.HBaseTestingUtil;
+import org.apache.hadoop.hbase.testclassification.ReplicationTests;
+import org.apache.hadoop.hbase.testclassification.SmallTests;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
+
+/**
+ * Unit tests for bulkload copy bandwidth throttling in {@link HFileReplicator} and
+ * {@link ReplicationSink}. Does not require a running HBase cluster.
+ */
+@Tag(ReplicationTests.TAG)
+@Tag(SmallTests.TAG)
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+public class TestHFileReplicatorBandwidth {
+
+  private final HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil();
+
+  @BeforeAll
+  public void setUpBeforeClass() throws Exception {
+    TEST_UTIL.startMiniDFSCluster(1);
+  }
+
+  @AfterAll
+  public void tearDownAfterClass() throws Exception {
+    TEST_UTIL.shutdownMiniDFSCluster();
+  }
+
+  /**
+   * Verify that onConfigurationChange updates the RateLimiter rate dynamically.
+   */
+  @Test
+  public void testBandwidthDynamicUpdate() throws Exception {
+    Configuration conf = TEST_UTIL.getConfiguration();
+    conf.set("hbase.replication.source.fs.conf.provider",
+      TestSourceFSConfigurationProvider.class.getCanonicalName());
+    ReplicationSink sink = new ReplicationSink(conf, null);
+
+    // Default: unlimited
+    assertEquals(Double.MAX_VALUE, sink.getBulkLoadCopyRateLimiterRate(), 0.0);
+
+    // Apply 50 MB/s
+    Configuration newConf = new Configuration(conf);
+    newConf.setDouble(HFileReplicator.REPLICATION_BULKLOAD_COPY_BANDWIDTH_MB_KEY, 50.0);
+    sink.onConfigurationChange(newConf);
+    assertEquals(50.0 * 1024 * 1024, sink.getBulkLoadCopyRateLimiterRate(), 1.0);
+
+    // Reset to unlimited (0 = no limit)
+    Configuration resetConf = new Configuration(conf);
+    resetConf.setDouble(HFileReplicator.REPLICATION_BULKLOAD_COPY_BANDWIDTH_MB_KEY, 0);
+    sink.onConfigurationChange(resetConf);
+    assertEquals(Double.MAX_VALUE, sink.getBulkLoadCopyRateLimiterRate(), 0.0);
+  }
+
+  /**
+   * Verify that throttled copy actually slows down I/O at the specified rate. Creates a file of
+   * known size, copies it through a throttled stream at 1 MB/s, and asserts the elapsed time is at
+   * least (fileSize / 1MB * 0.8) seconds.
+   */
+  @Test
+  public void testCopyIsThrottled() throws Exception {
+    Configuration conf = TEST_UTIL.getConfiguration();
+    FileSystem fs = TEST_UTIL.getTestFileSystem();
+    Path testDir = TEST_UTIL.getDataTestDirOnTestFS("testCopyIsThrottled");
+    fs.mkdirs(testDir);
+
+    // Write a ~2MB test file
+    Path srcFile = new Path(testDir, "src");
+    int fileSize = 2 * 1024 * 1024;
+    try (FSDataOutputStream out = fs.create(srcFile)) {
+      writeBytes(out, fileSize);
+    }
+
+    Path dstFile = new Path(testDir, "dst");
+
+    // Throttle at 1 MB/s
+    double limitMbPerSec = 1.0;
+    org.apache.hbase.thirdparty.com.google.common.util.concurrent.RateLimiter limiter =
+      org.apache.hbase.thirdparty.com.google.common.util.concurrent.RateLimiter
+        .create(limitMbPerSec * 1024 * 1024);
+
+    long start = System.currentTimeMillis();
+    try (java.io.InputStream in = fs.open(srcFile); OutputStream out = fs.create(dstFile)) {
+      byte[] buf = new byte[65536];
+      int bytesRead;
+      while ((bytesRead = in.read(buf)) >= 0) {
+        limiter.acquire(bytesRead);
+        out.write(buf, 0, bytesRead);
+      }
+    }
+    long elapsed = System.currentTimeMillis() - start;
+
+    // At 1MB/s, 2MB should take >= 1600ms (allow 20% margin for JVM overhead)
+    long minExpectedMs = (long) (fileSize * 1000L / (limitMbPerSec * 1024 * 1024) * 0.8);
+    assertTrue(elapsed >= minExpectedMs,
+      "Expected throttled copy >= " + minExpectedMs + "ms, got " + elapsed + "ms");
+  }
+
+  private static void writeBytes(OutputStream out, int size) throws IOException {
+    byte[] buf = new byte[65536];
+    int written = 0;
+    while (written < size) {
+      int chunk = Math.min(buf.length, size - written);
+      out.write(buf, 0, chunk);
+      written += chunk;
+    }
+  }
+}