HBASE-30238 HBase bulkload replication causes duplicate HFile loading and compaction storm when source RPC times out

hbase-server/src/main/java/org/apache/hadoop/hbase/master/HMaster.java

@@ -141,6 +141,7 @@
 import org.apache.hadoop.hbase.master.cleaner.HFileCleaner;
 import org.apache.hadoop.hbase.master.cleaner.LogCleaner;
 import org.apache.hadoop.hbase.master.cleaner.ReplicationBarrierCleaner;
+import org.apache.hadoop.hbase.master.cleaner.ReplicationBulkLoadEventCleaner;
 import org.apache.hadoop.hbase.master.cleaner.SnapshotCleanerChore;
 import org.apache.hadoop.hbase.master.hbck.HbckChore;
 import org.apache.hadoop.hbase.master.http.MasterDumpServlet;
@@ -435,6 +436,7 @@ public class HMaster extends HBaseServerBase<MasterRpcServices> implements Maste
   // The exclusive hfile cleaner pool for scanning the archive directory
   private DirScanPool exclusiveHFileCleanerPool;
   private ReplicationBarrierCleaner replicationBarrierCleaner;
+  private ReplicationBulkLoadEventCleaner replicationBulkLoadEventCleaner;
   private MobFileCleanerChore mobFileCleanerChore;
   private MobFileCompactionChore mobFileCompactionChore;
   private RollingUpgradeChore rollingUpgradeChore;
@@ -1795,6 +1797,9 @@ conf, getMasterFileSystem().getFileSystem(), new Path(archiveDir, path),
     replicationBarrierCleaner =
       new ReplicationBarrierCleaner(conf, this, getConnection(), replicationPeerManager);
     getChoreService().scheduleChore(replicationBarrierCleaner);
+    replicationBulkLoadEventCleaner =
+      new ReplicationBulkLoadEventCleaner(conf, this, getZooKeeper());
+    getChoreService().scheduleChore(replicationBulkLoadEventCleaner);
 
     final boolean isSnapshotChoreEnabled = this.snapshotCleanupStateStore.get();
     this.snapshotCleanerChore = new SnapshotCleanerChore(this, conf, getSnapshotManager());
@@ -1982,6 +1987,7 @@ protected void stopChores() {
       hfileCleaners = null;
     }
     shutdownChore(replicationBarrierCleaner);
+    shutdownChore(replicationBulkLoadEventCleaner);
     shutdownChore(snapshotCleanerChore);
     shutdownChore(hbckChore);
     shutdownChore(regionsRecoveryChore);

hbase-server/src/main/java/org/apache/hadoop/hbase/master/cleaner/ReplicationBulkLoadEventCleaner.java

@@ -0,0 +1,70 @@
+/*
+ * 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.master.cleaner;
+
+import java.util.concurrent.TimeUnit;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.ScheduledChore;
+import org.apache.hadoop.hbase.Stoppable;
+import org.apache.hadoop.hbase.replication.regionserver.ReplicationBulkLoadEventTracker;
+import org.apache.hadoop.hbase.zookeeper.ZKWatcher;
+import org.apache.yetus.audience.InterfaceAudience;
+import org.apache.zookeeper.KeeperException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Cleans completed replicated bulk load event markers after they are old enough.
+ */
+@InterfaceAudience.Private
+public class ReplicationBulkLoadEventCleaner extends ScheduledChore {
+
+  private static final Logger LOG = LoggerFactory.getLogger(ReplicationBulkLoadEventCleaner.class);
+
+  public static final String PERIOD_MS_KEY =
+    "hbase.master.cleaner.replication.bulkload.event.period.ms";
+  public static final int PERIOD_MS_DEFAULT = (int) TimeUnit.MINUTES.toMillis(10);
+  public static final String DONE_TTL_MS_KEY = "hbase.replication.bulkload.event.done.ttl.ms";
+  public static final long DONE_TTL_MS_DEFAULT = TimeUnit.DAYS.toMillis(1);
+
+  private final ReplicationBulkLoadEventTracker tracker;
+  private final long doneTtlMs;
+
+  public ReplicationBulkLoadEventCleaner(Configuration conf, Stoppable stopper, ZKWatcher zkw) {
+    super("ReplicationBulkLoadEventCleaner", stopper,
+      conf.getInt(PERIOD_MS_KEY, PERIOD_MS_DEFAULT));
+    this.tracker = new ReplicationBulkLoadEventTracker(conf, zkw);
+    this.doneTtlMs = conf.getLong(DONE_TTL_MS_KEY, DONE_TTL_MS_DEFAULT);
+  }
+
+  @Override
+  protected void chore() {
+    try {
+      int deleted = tracker.cleanDoneMarkers(doneTtlMs);
+      if (deleted > 0) {
+        LOG.info("Cleaned {} replicated bulkload event marker(s)", deleted);
+      }
+    } catch (KeeperException e) {
+      LOG.warn("Failed to clean replicated bulkload event markers", e);
+    }
+  }
+
+  public void choreForTesting() {
+    chore();
+  }
+}

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;
@@ -78,7 +79,7 @@ public void startReplicationService() throws IOException {
     if (server instanceof HRegionServer) {
       rsServerHost = ((HRegionServer) server).getRegionServerCoprocessorHost();
     }
-    this.replicationSink = new ReplicationSink(this.conf, rsServerHost);
+    this.replicationSink = new ReplicationSink(this.conf, rsServerHost, server.getZooKeeper());
     this.server.getChoreService().scheduleChore(new ReplicationStatisticsChore(
       "ReplicationSinkStatistics", server, (int) TimeUnit.SECONDS.toMillis(statsPeriodInSecond)));
   }
@@ -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,22 @@ 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. A low limit can make the
+   * sink-side bulkload copy exceed the replication RPC timeout, so configure the timeout
+   * accordingly.
+   */
+  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 +113,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 +142,7 @@ public HFileReplicator(Configuration sourceClusterConf, String sourceBaseNamespa
       REPLICATION_BULKLOAD_COPY_HFILES_PERTHREAD_DEFAULT);
 
     sinkFs = FileSystem.get(conf);
+    this.rateLimiter = rateLimiter;
   }
 
   @Override
@@ -336,16 +359,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 +378,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);
+        }
+      }
+    }
   }
 }