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

hbase-common/src/main/resources/hbase-default.xml

@@ -1867,6 +1867,79 @@ possible configurations would overwhelm and obscure the important.
         default of 10 will rarely need to be changed.
     </description>
   </property>
+  <property>
+    <name>hbase.replication.bulkload.copy.bandwidth.mb</name>
+    <value>0</value>
+    <description>
+      The maximum aggregate bandwidth, in megabytes per second, that a sink RegionServer uses while
+      copying HFiles from the source cluster for replicated bulk loads. 0 means no limit. This value
+      can be changed dynamically through configuration reload on the sink. If this is set too low,
+      the HFile copy can exceed the replication RPC timeout and cause the source to retry the same
+      batch.
+    </description>
+  </property>
+  <property>
+    <name>hbase.replication.bulkload.event.tracker.znode</name>
+    <value>bulkload-events</value>
+    <description>
+      The ZooKeeper znode name under the HBase replication znode used to coordinate replicated
+      bulk load events in the sink cluster. RegionServers create short-lived in-progress markers
+      and completed-event markers under this znode so that retries of the same replicated bulk
+      load WAL event are not executed more than once across different target RegionServers.
+    </description>
+  </property>
+  <property>
+    <name>hbase.replication.bulkload.event.bucket.width.ms</name>
+    <value>3600000</value>
+    <description>
+      The time width, in milliseconds, of each bucket used when storing replicated bulk load event
+      markers. The event write time is divided by this value to choose a bucket. Larger values
+      reduce the number of ZooKeeper bucket znodes but make each bucket contain more completed
+      event markers for the cleaner to scan. Smaller values create more bucket znodes but make
+      each bucket smaller.
+    </description>
+  </property>
+  <property>
+    <name>hbase.replication.bulkload.event.wait.timeout.ms</name>
+    <value>60000</value>
+    <description>
+      The maximum time, in milliseconds, that a sink RegionServer waits while another RegionServer
+      is processing the same replicated bulk load event. If the event is marked done before this
+      timeout, the retry is treated as already completed and skipped. If the timeout expires while
+      the in-progress marker still exists, replication fails the batch so the source can retry
+      later.
+    </description>
+  </property>
+  <property>
+    <name>hbase.replication.bulkload.event.wait.interval.ms</name>
+    <value>1000</value>
+    <description>
+      The sleep interval, in milliseconds, between checks while waiting for another sink
+      RegionServer to finish the same replicated bulk load event. Smaller values detect completion
+      sooner but issue more ZooKeeper checks. Larger values reduce ZooKeeper polling at the cost
+      of slower retry progress.
+    </description>
+  </property>
+  <property>
+    <name>hbase.master.cleaner.replication.bulkload.event.period.ms</name>
+    <value>600000</value>
+    <description>
+      The period, in milliseconds, at which the HBase Master runs the replicated bulk load event
+      marker cleaner. The cleaner removes expired completed-event markers after they are older
+      than hbase.replication.bulkload.event.done.ttl.ms and are no longer protected by a matching
+      in-progress marker.
+    </description>
+  </property>
+  <property>
+    <name>hbase.replication.bulkload.event.done.ttl.ms</name>
+    <value>86400000</value>
+    <description>
+      The minimum age, in milliseconds, before a completed replicated bulk load event marker can be
+      removed by the Master cleaner. This value should be long enough to cover expected replication
+      retry delays; if it is too small, a late retry may not find the completed marker and can
+      execute the same bulk load again. Larger values retain more ZooKeeper marker znodes.
+    </description>
+  </property>
   <!-- Static Web User Filter properties. -->
   <property>
     <name>hbase.http.staticuser.user</name>

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;
@@ -1803,6 +1805,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());
@@ -1990,6 +1995,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,67 @@
+/*
+ * 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.io.IOException;
+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.replication.regionserver.ZKReplicationBulkLoadEventTracker;
+import org.apache.hadoop.hbase.zookeeper.ZKWatcher;
+import org.apache.yetus.audience.InterfaceAudience;
+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 ZKReplicationBulkLoadEventTracker(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 (IOException e) {
+      LOG.warn("Failed to clean replicated bulkload event markers", e);
+    }
+  }
+}

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,32 @@ 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;
+  private static final BulkLoadTableLoadListener NO_OP_TABLE_LOAD_LISTENER =
+    new BulkLoadTableLoadListener() {
+      @Override
+      public void tableLoaded(String tableName) {
+      }
+
+      @Override
+      public void tableLoadFailed(String tableName) {
+      }
+    };
+  // Double.MAX_VALUE means no throttling.
+  private volatile RateLimiter rateLimiter;
 
   private Configuration sourceClusterConf;
   private String sourceBaseNamespaceDirPath;
@@ -99,6 +123,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 +152,7 @@ public HFileReplicator(Configuration sourceClusterConf, String sourceBaseNamespa
       REPLICATION_BULKLOAD_COPY_HFILES_PERTHREAD_DEFAULT);
 
     sinkFs = FileSystem.get(conf);
+    this.rateLimiter = rateLimiter;
   }
 
   @Override
@@ -129,7 +162,17 @@ public void close() throws IOException {
     }
   }
 
+  interface BulkLoadTableLoadListener {
+    void tableLoaded(String tableName) throws IOException;
+
+    void tableLoadFailed(String tableName);
+  }
+
   public Void replicate() throws IOException {
+    return replicate(NO_OP_TABLE_LOAD_LISTENER);
+  }
+
+  Void replicate(BulkLoadTableLoadListener tableLoadListener) throws IOException {
     // Copy all the hfiles to the local file system
     Map<String, Path> tableStagingDirsMap = copyHFilesToStagingDir();
 
@@ -151,10 +194,16 @@ public Void replicate() throws IOException {
       }
       fsDelegationToken.acquireDelegationToken(sinkFs);
       try {
-        doBulkLoad(conf, tableName, stagingDir, queue, maxRetries);
+        try {
+          doBulkLoad(conf, tableName, stagingDir, queue, maxRetries);
+        } catch (IOException e) {
+          tableLoadListener.tableLoadFailed(tableNameString);
+          throw e;
+        }
       } finally {
         cleanup(stagingDir);
       }
+      tableLoadListener.tableLoaded(tableNameString);
     }
     return null;
   }
@@ -336,16 +385,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 +404,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);
+        }
+      }
+    }
   }
 }