HDFS-17937. Limit fsimage transfer concurrency during checkpoint with multiple NameNodes.

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java

@@ -262,6 +262,9 @@ public class DFSConfigKeys extends CommonConfigurationKeys {
   public static final String  DFS_NAMENODE_CHECKPOINT_PARALLEL_UPLOAD_ENABLED_KEY =
       "dfs.namenode.checkpoint.parallel.upload.enabled";
   public static final boolean DFS_NAMENODE_CHECKPOINT_PARALLEL_UPLOAD_ENABLED_DEFAULT = false;
+  public static final String DFS_NAMENODE_CHECKPOINT_PARALLEL_UPLOAD_MAX_THREADS_KEY =
+      "dfs.namenode.checkpoint.parallel.upload.max-threads";
+  public static final int DFS_NAMENODE_CHECKPOINT_PARALLEL_UPLOAD_MAX_THREADS_DEFAULT = 0;
   public static final String  DFS_NAMENODE_MISSING_CHECKPOINT_PERIODS_BEFORE_SHUTDOWN_KEY = "dfs.namenode.missing.checkpoint.periods.before.shutdown";
   public static final int     DFS_NAMENODE_MISSING_CHECKPOINT_PERIODS_BEFORE_SHUTDOWN_DEFAULT = 3;
   public static final String  DFS_NAMENODE_HEARTBEAT_RECHECK_INTERVAL_KEY =

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/CheckpointConf.java

@@ -23,6 +23,7 @@
 import org.slf4j.LoggerFactory;
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.util.Preconditions;
 
 import org.apache.hadoop.thirdparty.com.google.common.collect.ImmutableList;
 
@@ -60,6 +61,11 @@ public class CheckpointConf {
    */
   private final boolean parallelUploadEnabled;
 
+  /**
+   * Max number of concurrent fsimage uploads from standby checkpointer.
+   */
+  private final int parallelUploadMaxThreads;
+
   public CheckpointConf(Configuration conf) {
     checkpointCheckPeriod = conf.getTimeDuration(
         DFS_NAMENODE_CHECKPOINT_CHECK_PERIOD_KEY,
@@ -77,6 +83,12 @@ public CheckpointConf(Configuration conf) {
     parallelUploadEnabled = conf.getBoolean(
         DFS_NAMENODE_CHECKPOINT_PARALLEL_UPLOAD_ENABLED_KEY,
         DFS_NAMENODE_CHECKPOINT_PARALLEL_UPLOAD_ENABLED_DEFAULT);
+    parallelUploadMaxThreads = conf.getInt(
+        DFS_NAMENODE_CHECKPOINT_PARALLEL_UPLOAD_MAX_THREADS_KEY,
+        DFS_NAMENODE_CHECKPOINT_PARALLEL_UPLOAD_MAX_THREADS_DEFAULT);
+    Preconditions.checkArgument(parallelUploadMaxThreads >= 0,
+        DFS_NAMENODE_CHECKPOINT_PARALLEL_UPLOAD_MAX_THREADS_KEY
+            + " must be greater than or equal to 0");
     warnForDeprecatedConfigs(conf);
   }
 
@@ -119,4 +131,8 @@ public double getQuietPeriod() {
   public boolean isParallelUploadEnabled() {
     return parallelUploadEnabled;
   }
+
+  public int getParallelUploadMaxThreads() {
+    return parallelUploadMaxThreads;
+  }
 }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/ha/StandbyCheckpointer.java

@@ -247,7 +247,18 @@ private void doCheckpoint() throws InterruptedException, IOException {
     // Do this in a separate thread to avoid blocking transition to active, but don't allow more
     // than the expected number of tasks to run or queue up
     // See HDFS-4816
-    int poolSize = checkpointConf.isParallelUploadEnabled() ? remoteNNAddresses.size() : 0;
+    // If parallel upload is abled, Parallel upload concurrency policy:
+    // - If maxThreads is 0, concurrency equals the number of remote NameNodes.
+    // - If maxThreads is greater than 0, concurrency is the smaller of
+    //   the remote Namenodes count and maxThreads.
+    int maxThreads = checkpointConf.getParallelUploadMaxThreads();
+    int poolSize = checkpointConf.isParallelUploadEnabled() ?
+        (maxThreads == 0 ?
+            remoteNNAddresses.size() :
+            Math.min(remoteNNAddresses.size(), maxThreads)) :
+        0;
+    LOG.info("Fsimage upload concurrency: {}, remote namenodes count: {}", poolSize,
+        remoteNNAddresses.size());
     ExecutorService executor =
         new ThreadPoolExecutor(poolSize, remoteNNAddresses.size(), 100, TimeUnit.MILLISECONDS,
             new LinkedBlockingQueue<>(remoteNNAddresses.size()), uploadThreadFactory);

hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml

@@ -1429,6 +1429,22 @@
   </description>
 </property>
 
+<property>
+  <name>dfs.namenode.checkpoint.parallel.upload.max-threads</name>
+  <value>0</value>
+  <description>
+    This config sets the maximum number of concurrent threads used to upload fsimage
+    from StandbyCheckpointer to other NameNodes.
+    It takes effect only when dfs.namenode.checkpoint.parallel.upload.enabled is true.
+    A smaller value reduces upload concurrency pressure on both the local standby.
+    If set to 0, the concurrency equals the number of remote NameNodes.
+    If greater than 0, the concurrency is the smaller of the remote NameNodes count and maxThreads.
+    Note: A lower max-threads value may slow fsimage uploads and extend the effective
+    checkpoint interval (dfs.namenode.checkpoint.period), tune this setting according to
+    your deployment environment.
+  </description>
+</property>
+
 <property>
   <name>dfs.namenode.checkpoint.check.quiet-multiplier</name>
   <value>1.5</value>

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/ha/TestStandbyCheckpoints.java

@@ -29,6 +29,7 @@
 import java.util.List;
 import java.util.Random;
 import java.util.function.Supplier;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
@@ -829,6 +830,60 @@ public void testPutFsimagePartFailed() throws Exception {
     assertTrue(snnCheckpointTime2 > snnCheckpointTime1);
   }
 
+  @Test
+  @Timeout(value = 300)
+  public void testCheckpointParallelUploadWithMaxThreads() throws Exception {
+    final int maxThreads = 1;
+    final AtomicInteger inFlightUploads = new AtomicInteger(0);
+    final AtomicInteger peakInFlightUploads = new AtomicInteger(0);
+
+    for (int i = 1; i < NUM_NNS; i++) {
+      cluster.getConfiguration(i)
+          .setBoolean(DFSConfigKeys.DFS_NAMENODE_CHECKPOINT_PARALLEL_UPLOAD_ENABLED_KEY, true);
+      cluster.getConfiguration(i)
+          .setInt(DFSConfigKeys.DFS_NAMENODE_CHECKPOINT_PARALLEL_UPLOAD_MAX_THREADS_KEY,
+              maxThreads);
+      cluster.getConfiguration(i).setLong(DFSConfigKeys.DFS_IMAGE_TRANSFER_RATE_KEY, 100);
+    }
+
+    CheckpointFaultInjector oldInjector = CheckpointFaultInjector.getInstance();
+    CheckpointFaultInjector.set(new CheckpointFaultInjector() {
+      @Override
+      public void duringUploadInProgess() throws InterruptedException {
+        int currentInFlight = inFlightUploads.incrementAndGet();
+        peakInFlightUploads.updateAndGet(previousPeak -> Math.max(previousPeak, currentInFlight));
+        try {
+          Thread.sleep(100);
+        } finally {
+          inFlightUploads.decrementAndGet();
+        }
+      }
+    });
+
+    try {
+      for (int i = 0; i < NUM_NNS; i++) {
+        cluster.restartNameNode(i);
+      }
+      setNNs();
+
+      cluster.transitionToActive(0);
+      doEdits(0, 100);
+
+      for (int i = 1; i < NUM_NNS; i++) {
+        HATestUtil.waitForStandbyToCatchUp(nns[0], nns[i]);
+        HATestUtil.waitForCheckpoint(cluster, i, ImmutableList.of(104));
+      }
+
+      assertTrue(peakInFlightUploads.get() > 0,
+          "Expected at least one fsimage upload to be observed.");
+      assertTrue(peakInFlightUploads.get() <= maxThreads,
+          "Observed upload concurrency " + peakInFlightUploads.get()
+              + " exceeds configured max threads " + maxThreads);
+    } finally {
+      CheckpointFaultInjector.set(oldInjector);
+    }
+  }
+
   /**
    * A codec which just slows down the saving of the image significantly
    * by sleeping a few milliseconds on every write. This makes it easy to