diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java
index 2fb2de5746fafa..214b11a43936fe 100755
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java
+++ b/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 =
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/CheckpointConf.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/CheckpointConf.java
index a5f8049c95066e..65534384a2f0e2 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/CheckpointConf.java
+++ b/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;
+ }
}
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/ha/StandbyCheckpointer.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/ha/StandbyCheckpointer.java
index 0e8c067322ba3d..5c91ee5eb52fa9 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/ha/StandbyCheckpointer.java
+++ b/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);
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml b/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml
index 1c56231fe26c61..b9121ef886ac65 100755
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml
@@ -1429,6 +1429,22 @@
+
+ dfs.namenode.checkpoint.parallel.upload.max-threads
+ 0
+
+ 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.
+
+
+
dfs.namenode.checkpoint.check.quiet-multiplier
1.5
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/ha/TestStandbyCheckpoints.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/ha/TestStandbyCheckpoints.java
index 1b03ec946893b9..c6b49590bac55c 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/ha/TestStandbyCheckpoints.java
+++ b/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